tdatablock.c 57.3 KB
Newer Older
S
Shengliang Guan 已提交
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/>.
 */

#define _DEFAULT_SOURCE
H
Haojun Liao 已提交
17
#include "tdatablock.h"
S
compare  
Shengliang Guan 已提交
18
#include "tcompare.h"
19
#include "tglobal.h"
20
#include "tlog.h"
21
#include "tname.h"
22

L
Liu Jicong 已提交
23
int32_t taosGetFqdnPortFromEp(const char* ep, SEp* pEp) {
H
Haojun Liao 已提交
24 25
  pEp->port = 0;
  strcpy(pEp->fqdn, ep);
26

L
Liu Jicong 已提交
27
  char* temp = strchr(pEp->fqdn, ':');
28 29
  if (temp) {
    *temp = 0;
L
Liu Jicong 已提交
30
    pEp->port = atoi(temp + 1);
31 32
  }

H
Haojun Liao 已提交
33
  if (pEp->port == 0) {
S
Shengliang Guan 已提交
34
    pEp->port = tsServerPort;
35 36 37 38 39
  }

  return 0;
}

L
Liu Jicong 已提交
40
void addEpIntoEpSet(SEpSet* pEpSet, const char* fqdn, uint16_t port) {
H
Haojun Liao 已提交
41 42 43 44 45 46 47 48 49 50
  if (pEpSet == NULL || fqdn == NULL || strlen(fqdn) == 0) {
    return;
  }

  int32_t index = pEpSet->numOfEps;
  tstrncpy(pEpSet->eps[index].fqdn, fqdn, tListLen(pEpSet->eps[index].fqdn));
  pEpSet->eps[index].port = port;
  pEpSet->numOfEps += 1;
}

L
Liu Jicong 已提交
51
bool isEpsetEqual(const SEpSet* s1, const SEpSet* s2) {
52 53 54 55 56
  if (s1->numOfEps != s2->numOfEps || s1->inUse != s2->inUse) {
    return false;
  }

  for (int32_t i = 0; i < s1->numOfEps; i++) {
L
Liu Jicong 已提交
57
    if (s1->eps[i].port != s2->eps[i].port || strncmp(s1->eps[i].fqdn, s2->eps[i].fqdn, TSDB_FQDN_LEN) != 0)
58 59 60 61 62
      return false;
  }
  return true;
}

L
Liu Jicong 已提交
63
void updateEpSet_s(SCorEpSet* pEpSet, SEpSet* pNewEpSet) {
64 65 66 67 68
  taosCorBeginWrite(&pEpSet->version);
  pEpSet->epSet = *pNewEpSet;
  taosCorEndWrite(&pEpSet->version);
}

L
Liu Jicong 已提交
69
SEpSet getEpSet_s(SCorEpSet* pEpSet) {
70 71 72 73 74 75 76 77
  SEpSet ep = {0};
  taosCorBeginRead(&pEpSet->version);
  ep = pEpSet->epSet;
  taosCorEndRead(&pEpSet->version);

  return ep;
}

78
int32_t colDataGetLength(const SColumnInfoData* pColumnInfoData, int32_t numOfRows) {
H
Haojun Liao 已提交
79 80 81 82
  ASSERT(pColumnInfoData != NULL);
  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
    return pColumnInfoData->varmeta.length;
  } else {
83 84 85 86 87
    if (pColumnInfoData->info.type == TSDB_DATA_TYPE_NULL) {
      return 0;
    } else {
      return pColumnInfoData->info.bytes * numOfRows;
    }
H
Haojun Liao 已提交
88 89 90
  }
}

H
Haojun Liao 已提交
91 92 93 94 95 96 97 98
int32_t colDataGetFullLength(const SColumnInfoData* pColumnInfoData, int32_t numOfRows) {
  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
    return pColumnInfoData->varmeta.length + sizeof(int32_t) * numOfRows;
  } else {
    return pColumnInfoData->info.bytes * numOfRows + BitmapLen(numOfRows);
  }
}

H
Haojun Liao 已提交
99 100 101 102
void colDataTrim(SColumnInfoData* pColumnInfoData) {
  // TODO
}

103
int32_t getJsonValueLen(const char* data) {
wmmhello's avatar
wmmhello 已提交
104 105 106 107 108 109 110 111 112
  int32_t dataLen = 0;
  if (*data == TSDB_DATA_TYPE_NULL) {
    dataLen = CHAR_BYTES;
  } else if (*data == TSDB_DATA_TYPE_NCHAR) {
    dataLen = varDataTLen(data + CHAR_BYTES) + CHAR_BYTES;
  } else if (*data == TSDB_DATA_TYPE_DOUBLE) {
    dataLen = DOUBLE_BYTES + CHAR_BYTES;
  } else if (*data == TSDB_DATA_TYPE_BOOL) {
    dataLen = CHAR_BYTES + CHAR_BYTES;
113
  } else if (*data & TD_TAG_JSON) {  // json string
wmmhello's avatar
wmmhello 已提交
114 115 116 117 118 119 120
    dataLen = ((STag*)(data))->len;
  } else {
    ASSERT(0);
  }
  return dataLen;
}

121 122 123 124
int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull) {
  ASSERT(pColumnInfoData != NULL);

  if (isNull) {
H
Haojun Liao 已提交
125 126
    // There is a placehold for each NULL value of binary or nchar type.
    if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
L
Liu Jicong 已提交
127
      pColumnInfoData->varmeta.offset[currentRow] = -1;  // it is a null value of VAR type.
H
Haojun Liao 已提交
128 129 130 131
    } else {
      colDataSetNull_f(pColumnInfoData->nullbitmap, currentRow);
    }

H
Haojun Liao 已提交
132
    pColumnInfoData->hasNull = true;
133 134 135 136 137
    return 0;
  }

  int32_t type = pColumnInfoData->info.type;
  if (IS_VAR_DATA_TYPE(type)) {
wmmhello's avatar
wmmhello 已提交
138
    int32_t dataLen = 0;
139
    if (type == TSDB_DATA_TYPE_JSON) {
wmmhello's avatar
wmmhello 已提交
140
      dataLen = getJsonValueLen(pData);
141
    } else {
wmmhello's avatar
wmmhello 已提交
142
      dataLen = varDataTLen(pData);
143
    }
144

H
Haojun Liao 已提交
145
    SVarColAttr* pAttr = &pColumnInfoData->varmeta;
146
    if (pAttr->allocLen < pAttr->length + dataLen) {
H
Haojun Liao 已提交
147
      uint32_t newSize = pAttr->allocLen;
148
      if (newSize <= 1) {
H
Haojun Liao 已提交
149 150 151
        newSize = 8;
      }

152
      while (newSize < pAttr->length + dataLen) {
H
Haojun Liao 已提交
153 154 155
        newSize = newSize * 1.5;
      }

wafwerar's avatar
wafwerar 已提交
156
      char* buf = taosMemoryRealloc(pColumnInfoData->pData, newSize);
H
Haojun Liao 已提交
157
      if (buf == NULL) {
H
Haojun Liao 已提交
158
        return TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
159 160 161 162 163 164 165 166 167
      }

      pColumnInfoData->pData = buf;
      pAttr->allocLen = newSize;
    }

    uint32_t len = pColumnInfoData->varmeta.length;
    pColumnInfoData->varmeta.offset[currentRow] = len;

168 169
    memcpy(pColumnInfoData->pData + len, pData, dataLen);
    pColumnInfoData->varmeta.length += dataLen;
170
  } else {
wmmhello's avatar
wmmhello 已提交
171
    memcpy(pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow, pData, pColumnInfoData->info.bytes);
172 173 174 175 176
  }

  return 0;
}

L
Liu Jicong 已提交
177 178
static void doBitmapMerge(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, const SColumnInfoData* pSource,
                          int32_t numOfRow2) {
wmmhello's avatar
wmmhello 已提交
179 180
  if (numOfRow2 <= 0) return;

H
Haojun Liao 已提交
181 182 183 184 185 186 187
  uint32_t total = numOfRow1 + numOfRow2;

  uint32_t remindBits = BitPos(numOfRow1);
  uint32_t shiftBits = 8 - remindBits;

  if (remindBits == 0) {  // no need to shift bits of bitmap
    memcpy(pColumnInfoData->nullbitmap + BitmapLen(numOfRow1), pSource->nullbitmap, BitmapLen(numOfRow2));
wmmhello's avatar
wmmhello 已提交
188 189
    return;
  }
H
Haojun Liao 已提交
190

wmmhello's avatar
wmmhello 已提交
191 192
  uint8_t* p = (uint8_t*)pSource->nullbitmap;
  pColumnInfoData->nullbitmap[BitmapLen(numOfRow1) - 1] |= (p[0] >> remindBits);  // copy remind bits
H
Haojun Liao 已提交
193

wmmhello's avatar
wmmhello 已提交
194 195 196
  if (BitmapLen(numOfRow1) == BitmapLen(total)) {
    return;
  }
H
Haojun Liao 已提交
197

wmmhello's avatar
wmmhello 已提交
198 199
  int32_t len = BitmapLen(numOfRow2);
  int32_t i = 0;
H
Haojun Liao 已提交
200

wmmhello's avatar
wmmhello 已提交
201
  uint8_t* start = (uint8_t*)&pColumnInfoData->nullbitmap[BitmapLen(numOfRow1)];
C
Cary Xu 已提交
202 203
  int32_t  overCount = BitmapLen(total) - BitmapLen(numOfRow1);
  while (i < len) {  // size limit of pSource->nullbitmap
wmmhello's avatar
wmmhello 已提交
204
    if (i >= 1) {
C
Cary Xu 已提交
205
      start[i - 1] |= (p[i] >> remindBits);  // copy remind bits
H
Haojun Liao 已提交
206
    }
wmmhello's avatar
wmmhello 已提交
207

C
Cary Xu 已提交
208
    if (i >= overCount) {  // size limit of pColumnInfoData->nullbitmap
wmmhello's avatar
wmmhello 已提交
209 210 211
      return;
    }

C
Cary Xu 已提交
212
    start[i] |= (p[i] << shiftBits);  // copy shift bits
wmmhello's avatar
wmmhello 已提交
213
    i += 1;
H
Haojun Liao 已提交
214 215 216
  }
}

217 218
int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, int32_t* capacity,
                        const SColumnInfoData* pSource, uint32_t numOfRow2) {
H
Haojun Liao 已提交
219 220 221 222 223
  ASSERT(pColumnInfoData != NULL && pSource != NULL && pColumnInfoData->info.type == pSource->info.type);
  if (numOfRow2 == 0) {
    return numOfRow1;
  }

wmmhello's avatar
wmmhello 已提交
224 225 226
  if (pSource->hasNull) {
    pColumnInfoData->hasNull = pSource->hasNull;
  }
227

228
  uint32_t finalNumOfRows = numOfRow1 + numOfRow2;
H
Haojun Liao 已提交
229 230
  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
    // Handle the bitmap
231 232 233 234 235 236 237 238
    if (finalNumOfRows > *capacity) {
      char* p = taosMemoryRealloc(pColumnInfoData->varmeta.offset, sizeof(int32_t) * (numOfRow1 + numOfRow2));
      if (p == NULL) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }

      *capacity = finalNumOfRows;
      pColumnInfoData->varmeta.offset = (int32_t*)p;
H
Haojun Liao 已提交
239 240
    }

L
Liu Jicong 已提交
241
    for (int32_t i = 0; i < numOfRow2; ++i) {
242 243 244 245 246
      if (pSource->varmeta.offset[i] == -1) {
        pColumnInfoData->varmeta.offset[i + numOfRow1] = -1;
      } else {
        pColumnInfoData->varmeta.offset[i + numOfRow1] = pSource->varmeta.offset[i] + pColumnInfoData->varmeta.length;
      }
247
    }
H
Haojun Liao 已提交
248

249
    // copy data
H
Haojun Liao 已提交
250 251 252
    uint32_t len = pSource->varmeta.length;
    uint32_t oldLen = pColumnInfoData->varmeta.length;
    if (pColumnInfoData->varmeta.allocLen < len + oldLen) {
wafwerar's avatar
wafwerar 已提交
253
      char* tmp = taosMemoryRealloc(pColumnInfoData->pData, len + oldLen);
H
Haojun Liao 已提交
254
      if (tmp == NULL) {
255
        return TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
256 257 258 259 260 261
      }

      pColumnInfoData->pData = tmp;
      pColumnInfoData->varmeta.allocLen = len + oldLen;
    }

262 263
    memcpy(pColumnInfoData->pData + oldLen, pSource->pData, len);
    pColumnInfoData->varmeta.length = len + oldLen;
H
Haojun Liao 已提交
264
  } else {
265
    if (finalNumOfRows > *capacity) {
266
      char* tmp = taosMemoryRealloc(pColumnInfoData->pData, finalNumOfRows * pColumnInfoData->info.bytes);
267 268 269
      if (tmp == NULL) {
        return TSDB_CODE_VND_OUT_OF_MEMORY;
      }
H
Haojun Liao 已提交
270

271 272 273 274 275 276 277 278 279 280 281
      pColumnInfoData->pData = tmp;

      if (BitmapLen(numOfRow1) < BitmapLen(finalNumOfRows)) {
        char*    btmp = taosMemoryRealloc(pColumnInfoData->nullbitmap, BitmapLen(finalNumOfRows));
        uint32_t extend = BitmapLen(finalNumOfRows) - BitmapLen(numOfRow1);
        memset(btmp + BitmapLen(numOfRow1), 0, extend);

        pColumnInfoData->nullbitmap = btmp;
      }

      *capacity = finalNumOfRows;
H
Haojun Liao 已提交
282 283
    }

284 285
    doBitmapMerge(pColumnInfoData, numOfRow1, pSource, numOfRow2);

X
Xiaoyu Wang 已提交
286 287 288 289
    if (pSource->pData) {
      int32_t offset = pColumnInfoData->info.bytes * numOfRow1;
      memcpy(pColumnInfoData->pData + offset, pSource->pData, pSource->info.bytes * numOfRow2);
    }
H
Haojun Liao 已提交
290 291 292 293 294
  }

  return numOfRow1 + numOfRow2;
}

295 296
int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows) {
  ASSERT(pColumnInfoData != NULL && pSource != NULL && pColumnInfoData->info.type == pSource->info.type);
297
  if (numOfRows <= 0) {
298 299 300 301 302
    return numOfRows;
  }

  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
    // Handle the bitmap
wafwerar's avatar
wafwerar 已提交
303
    char* p = taosMemoryRealloc(pColumnInfoData->varmeta.offset, sizeof(int32_t) * numOfRows);
304 305 306 307
    if (p == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }

L
Liu Jicong 已提交
308
    pColumnInfoData->varmeta.offset = (int32_t*)p;
309 310 311
    memcpy(pColumnInfoData->varmeta.offset, pSource->varmeta.offset, sizeof(int32_t) * numOfRows);

    if (pColumnInfoData->varmeta.allocLen < pSource->varmeta.length) {
wafwerar's avatar
wafwerar 已提交
312
      char* tmp = taosMemoryRealloc(pColumnInfoData->pData, pSource->varmeta.length);
313 314 315 316 317 318 319 320 321 322 323
      if (tmp == NULL) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }

      pColumnInfoData->pData = tmp;
      pColumnInfoData->varmeta.allocLen = pSource->varmeta.length;
    }

    memcpy(pColumnInfoData->pData, pSource->pData, pSource->varmeta.length);
    pColumnInfoData->varmeta.length = pSource->varmeta.length;
  } else {
wafwerar's avatar
wafwerar 已提交
324
    char* tmp = taosMemoryRealloc(pColumnInfoData->nullbitmap, BitmapLen(numOfRows));
325 326 327 328 329 330 331
    if (tmp == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }

    pColumnInfoData->nullbitmap = tmp;
    memcpy(pColumnInfoData->nullbitmap, pSource->nullbitmap, BitmapLen(numOfRows));

X
Xiaoyu Wang 已提交
332
    if (pSource->pData) {
X
Xiaoyu Wang 已提交
333 334 335 336 337
      int32_t newSize = numOfRows * pColumnInfoData->info.bytes;
      tmp = taosMemoryRealloc(pColumnInfoData->pData, newSize);
      if (tmp == NULL) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }
338

X
Xiaoyu Wang 已提交
339 340 341
      pColumnInfoData->pData = tmp;
      memcpy(pColumnInfoData->pData, pSource->pData, pSource->info.bytes * numOfRows);
    }
342 343
  }

344 345
  pColumnInfoData->hasNull = pSource->hasNull;
  pColumnInfoData->info = pSource->info;
346 347 348
  return 0;
}

H
Haojun Liao 已提交
349
size_t blockDataGetNumOfCols(const SSDataBlock* pBlock) {
H
Haojun Liao 已提交
350
  ASSERT(pBlock && pBlock->info.numOfCols == taosArrayGetSize(pBlock->pDataBlock));
351 352 353
  return pBlock->info.numOfCols;
}

L
Liu Jicong 已提交
354
size_t blockDataGetNumOfRows(const SSDataBlock* pBlock) { return pBlock->info.rows; }
355

356
int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock, int32_t tsColumnIndex) {
357 358 359 360 361 362 363 364
  if (pDataBlock == NULL || pDataBlock->info.rows <= 0) {
    return 0;
  }

  if (pDataBlock->info.numOfCols <= 0) {
    return -1;
  }

365 366
  int32_t index = (tsColumnIndex == -1) ? 0 : tsColumnIndex;

367
  SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, index);
368 369 370 371
  if (pColInfoData->info.type != TSDB_DATA_TYPE_TIMESTAMP) {
    return 0;
  }

372 373 374 375 376 377
  TSKEY skey = *(TSKEY*)colDataGetData(pColInfoData, 0);
  TSKEY ekey = *(TSKEY*)colDataGetData(pColInfoData, (pDataBlock->info.rows - 1));

  pDataBlock->info.window.skey = TMIN(skey, ekey);
  pDataBlock->info.window.ekey = TMAX(skey, ekey);

378 379 380
  return 0;
}

381
int32_t blockDataMerge(SSDataBlock* pDest, const SSDataBlock* pSrc) {
wmmhello's avatar
wmmhello 已提交
382
  assert(pSrc != NULL && pDest != NULL);
383 384
  int32_t capacity = pDest->info.capacity;

385
  for (int32_t i = 0; i < pDest->info.numOfCols; ++i) {
H
Haojun Liao 已提交
386
    SColumnInfoData* pCol2 = taosArrayGet(pDest->pDataBlock, i);
387
    SColumnInfoData* pCol1 = taosArrayGet(pSrc->pDataBlock, i);
H
Haojun Liao 已提交
388

389 390
    capacity = pDest->info.capacity;
    colDataMergeCol(pCol2, pDest->info.rows, &capacity, pCol1, pSrc->info.rows);
H
Haojun Liao 已提交
391 392
  }

393
  pDest->info.capacity = capacity;
H
Haojun Liao 已提交
394 395 396 397 398 399 400
  pDest->info.rows += pSrc->info.rows;
  return TSDB_CODE_SUCCESS;
}

size_t blockDataGetSize(const SSDataBlock* pBlock) {
  assert(pBlock != NULL);

L
Liu Jicong 已提交
401
  size_t  total = 0;
H
Haojun Liao 已提交
402
  int32_t numOfCols = pBlock->info.numOfCols;
H
Haojun Liao 已提交
403

L
Liu Jicong 已提交
404
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
405
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
H
Haojun Liao 已提交
406
    total += colDataGetFullLength(pColInfoData, pBlock->info.rows);
H
Haojun Liao 已提交
407 408 409 410 411 412 413
  }

  return total;
}

// the number of tuples can be fit in one page.
// Actual data rows pluses the corresponding meta data must fit in one memory buffer of the given page size.
L
Liu Jicong 已提交
414 415
int32_t blockDataSplitRows(SSDataBlock* pBlock, bool hasVarCol, int32_t startIndex, int32_t* stopIndex,
                           int32_t pageSize) {
H
Haojun Liao 已提交
416 417 418 419 420
  ASSERT(pBlock != NULL && stopIndex != NULL);

  int32_t numOfCols = pBlock->info.numOfCols;
  int32_t numOfRows = pBlock->info.rows;

H
Haojun Liao 已提交
421 422
  int32_t bitmapChar = 1;

L
Liu Jicong 已提交
423
  size_t headerSize = sizeof(int32_t);
424
  size_t colHeaderSize = sizeof(int32_t) * numOfCols;
L
Liu Jicong 已提交
425
  size_t payloadSize = pageSize - (headerSize + colHeaderSize);
426

H
Haojun Liao 已提交
427
  // TODO speedup by checking if the whole page can fit in firstly.
H
Haojun Liao 已提交
428
  if (!hasVarCol) {
L
Liu Jicong 已提交
429
    size_t  rowSize = blockDataGetRowSize(pBlock);
430 431
    int32_t capacity = payloadSize / (rowSize + numOfCols * bitmapChar / 8.0);
    ASSERT(capacity > 0);
432

433
    *stopIndex = startIndex + capacity - 1;
H
Haojun Liao 已提交
434 435 436
    if (*stopIndex >= numOfRows) {
      *stopIndex = numOfRows - 1;
    }
437

H
Haojun Liao 已提交
438
    return TSDB_CODE_SUCCESS;
439
  }
wmmhello's avatar
wmmhello 已提交
440 441 442 443 444 445
  // iterate the rows that can be fit in this buffer page
  int32_t size = (headerSize + colHeaderSize);
  for (int32_t j = startIndex; j < numOfRows; ++j) {
    for (int32_t i = 0; i < numOfCols; ++i) {
      SColumnInfoData* pColInfoData = TARRAY_GET_ELEM(pBlock->pDataBlock, i);
      if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
446
        if (pColInfoData->varmeta.offset[j] != -1) {
wmmhello's avatar
wmmhello 已提交
447 448
          char* p = colDataGetData(pColInfoData, j);
          size += varDataTLen(p);
H
Haojun Liao 已提交
449 450
        }

wmmhello's avatar
wmmhello 已提交
451
        size += sizeof(pColInfoData->varmeta.offset[0]);
452
      } else {
wmmhello's avatar
wmmhello 已提交
453
        size += pColInfoData->info.bytes;
H
Haojun Liao 已提交
454

wmmhello's avatar
wmmhello 已提交
455 456 457
        if (((j - startIndex) & 0x07) == 0) {
          size += 1;  // the space for null bitmap
        }
H
Haojun Liao 已提交
458 459 460
      }
    }

wmmhello's avatar
wmmhello 已提交
461
    if (size > pageSize) {  // pageSize must be able to hold one row
wmmhello's avatar
wmmhello 已提交
462
      *stopIndex = j - 1;
wmmhello's avatar
wmmhello 已提交
463
      ASSERT(*stopIndex >= startIndex);
wmmhello's avatar
wmmhello 已提交
464 465 466

      return TSDB_CODE_SUCCESS;
    }
H
Haojun Liao 已提交
467
  }
wmmhello's avatar
wmmhello 已提交
468 469 470 471

  // all fit in
  *stopIndex = numOfRows - 1;
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
472 473
}

474 475 476 477 478
SSDataBlock* blockDataExtractBlock(SSDataBlock* pBlock, int32_t startIndex, int32_t rowCount) {
  if (pBlock == NULL || startIndex < 0 || rowCount > pBlock->info.rows || rowCount + startIndex > pBlock->info.rows) {
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
479
  SSDataBlock* pDst = taosMemoryCalloc(1, sizeof(SSDataBlock));
480 481 482 483
  if (pDst == NULL) {
    return NULL;
  }

484 485 486 487 488
  pDst->info = pBlock->info;

  pDst->info.rows = 0;
  pDst->pDataBlock = taosArrayInit(pBlock->info.numOfCols, sizeof(SColumnInfoData));

L
Liu Jicong 已提交
489 490
  for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
    SColumnInfoData  colInfo = {0};
491 492 493 494 495
    SColumnInfoData* pSrcCol = taosArrayGet(pBlock->pDataBlock, i);
    colInfo.info = pSrcCol->info;

    if (IS_VAR_DATA_TYPE(pSrcCol->info.type)) {
      SVarColAttr* pAttr = &colInfo.varmeta;
wafwerar's avatar
wafwerar 已提交
496
      pAttr->offset = taosMemoryCalloc(rowCount, sizeof(int32_t));
497
    } else {
wafwerar's avatar
wafwerar 已提交
498 499
      colInfo.nullbitmap = taosMemoryCalloc(1, BitmapLen(rowCount));
      colInfo.pData = taosMemoryCalloc(rowCount, colInfo.info.bytes);
500 501 502 503 504 505 506 507 508 509
    }

    taosArrayPush(pDst->pDataBlock, &colInfo);
  }

  for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
    SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
    SColumnInfoData* pDstCol = taosArrayGet(pDst->pDataBlock, i);

    for (int32_t j = startIndex; j < (startIndex + rowCount); ++j) {
510 511
      bool isNull = false;
      if (pBlock->pBlockAgg == NULL) {
512
        isNull = colDataIsNull_s(pColData, j);
513 514 515
      } else {
        isNull = colDataIsNull(pColData, pBlock->info.rows, j, pBlock->pBlockAgg[i]);
      }
516

517
      char* p = colDataGetData(pColData, j);
518 519 520 521 522 523 524 525
      colDataAppend(pDstCol, j - startIndex, p, isNull);
    }
  }

  pDst->info.rows = rowCount;
  return pDst;
}

H
Haojun Liao 已提交
526 527
/**
 *
528 529 530 531 532
 * +------------------+---------------------------------------------+
 * |the number of rows|                    column #1                |
 * |    (4 bytes)     |------------+-----------------------+--------+
 * |                  | null bitmap| column length(4bytes) | values |
 * +------------------+------------+-----------------------+--------+
H
Haojun Liao 已提交
533 534 535 536
 * @param buf
 * @param pBlock
 * @return
 */
537
int32_t blockDataToBuf(char* buf, const SSDataBlock* pBlock) {
H
Haojun Liao 已提交
538 539 540
  ASSERT(pBlock != NULL);

  // write the number of rows
L
Liu Jicong 已提交
541
  *(uint32_t*)buf = pBlock->info.rows;
H
Haojun Liao 已提交
542 543 544 545 546 547

  int32_t numOfCols = pBlock->info.numOfCols;
  int32_t numOfRows = pBlock->info.rows;

  char* pStart = buf + sizeof(uint32_t);

L
Liu Jicong 已提交
548
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
549 550 551 552 553 554 555 556 557
    SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);
    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
      memcpy(pStart, pCol->varmeta.offset, numOfRows * sizeof(int32_t));
      pStart += numOfRows * sizeof(int32_t);
    } else {
      memcpy(pStart, pCol->nullbitmap, BitmapLen(numOfRows));
      pStart += BitmapLen(pBlock->info.rows);
    }

558
    uint32_t dataSize = colDataGetLength(pCol, numOfRows);
559

L
Liu Jicong 已提交
560
    *(int32_t*)pStart = dataSize;
561 562
    pStart += sizeof(int32_t);

H
Haojun Liao 已提交
563 564 565 566 567 568 569
    memcpy(pStart, pCol->pData, dataSize);
    pStart += dataSize;
  }

  return 0;
}

570
int32_t blockDataFromBuf(SSDataBlock* pBlock, const char* buf) {
L
Liu Jicong 已提交
571
  pBlock->info.rows = *(int32_t*)buf;
572

L
Liu Jicong 已提交
573
  int32_t     numOfCols = pBlock->info.numOfCols;
574 575
  const char* pStart = buf + sizeof(uint32_t);

L
Liu Jicong 已提交
576
  for (int32_t i = 0; i < numOfCols; ++i) {
577 578 579
    SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);

    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
H
Haojun Liao 已提交
580
      size_t metaSize = pBlock->info.rows * sizeof(int32_t);
C
Cary Xu 已提交
581
      char*  tmp = taosMemoryRealloc(pCol->varmeta.offset, metaSize);  // preview calloc is too small
wmmhello's avatar
wmmhello 已提交
582 583 584 585
      if (tmp == NULL) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }
      pCol->varmeta.offset = (int32_t*)tmp;
586 587 588 589 590 591 592
      memcpy(pCol->varmeta.offset, pStart, metaSize);
      pStart += metaSize;
    } else {
      memcpy(pCol->nullbitmap, pStart, BitmapLen(pBlock->info.rows));
      pStart += BitmapLen(pBlock->info.rows);
    }

L
Liu Jicong 已提交
593
    int32_t colLength = *(int32_t*)pStart;
594 595
    pStart += sizeof(int32_t);

596 597
    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
      if (pCol->varmeta.allocLen < colLength) {
wafwerar's avatar
wafwerar 已提交
598
        char* tmp = taosMemoryRealloc(pCol->pData, colLength);
599 600 601 602 603 604 605 606 607 608
        if (tmp == NULL) {
          return TSDB_CODE_OUT_OF_MEMORY;
        }

        pCol->pData = tmp;
        pCol->varmeta.allocLen = colLength;
      }

      pCol->varmeta.length = colLength;
      ASSERT(pCol->varmeta.length <= pCol->varmeta.allocLen);
609 610 611 612 613
    }

    memcpy(pCol->pData, pStart, colLength);
    pStart += colLength;
  }
614 615

  return TSDB_CODE_SUCCESS;
616 617
}

H
Haojun Liao 已提交
618
int32_t blockDataFromBuf1(SSDataBlock* pBlock, const char* buf, size_t capacity) {
X
Xiaoyu Wang 已提交
619 620
  pBlock->info.rows = *(int32_t*)buf;
  pBlock->info.groupId = *(uint64_t*)(buf + sizeof(int32_t));
H
Haojun Liao 已提交
621 622

  int32_t     numOfCols = pBlock->info.numOfCols;
623
  const char* pStart = buf + sizeof(uint32_t) + sizeof(uint64_t);
H
Haojun Liao 已提交
624 625 626

  for (int32_t i = 0; i < numOfCols; ++i) {
    SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);
627
    pCol->hasNull = true;
H
Haojun Liao 已提交
628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656

    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
      size_t metaSize = capacity * sizeof(int32_t);
      memcpy(pCol->varmeta.offset, pStart, metaSize);
      pStart += metaSize;
    } else {
      memcpy(pCol->nullbitmap, pStart, BitmapLen(capacity));
      pStart += BitmapLen(capacity);
    }

    int32_t colLength = *(int32_t*)pStart;
    pStart += sizeof(int32_t);

    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
      if (pCol->varmeta.allocLen < colLength) {
        char* tmp = taosMemoryRealloc(pCol->pData, colLength);
        if (tmp == NULL) {
          return TSDB_CODE_OUT_OF_MEMORY;
        }

        pCol->pData = tmp;
        pCol->varmeta.allocLen = colLength;
      }

      pCol->varmeta.length = colLength;
      ASSERT(pCol->varmeta.length <= pCol->varmeta.allocLen);
    }

    memcpy(pCol->pData, pStart, colLength);
657
    pStart += pCol->info.bytes * capacity;
H
Haojun Liao 已提交
658 659 660 661 662
  }

  return TSDB_CODE_SUCCESS;
}

663
size_t blockDataGetRowSize(SSDataBlock* pBlock) {
H
Haojun Liao 已提交
664
  ASSERT(pBlock != NULL);
665 666
  if (pBlock->info.rowSize == 0) {
    size_t rowSize = 0;
H
Haojun Liao 已提交
667

L
Liu Jicong 已提交
668 669 670 671 672
    size_t numOfCols = pBlock->info.numOfCols;
    for (int32_t i = 0; i < numOfCols; ++i) {
      SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
      rowSize += pColInfo->info.bytes;
    }
673 674

    pBlock->info.rowSize = rowSize;
H
Haojun Liao 已提交
675 676
  }

677
  return pBlock->info.rowSize;
H
Haojun Liao 已提交
678 679
}

H
Haojun Liao 已提交
680 681 682 683 684
/**
 * @refitem blockDataToBuf for the meta size
 * @param pBlock
 * @return
 */
685 686
size_t blockDataGetSerialMetaSize(uint32_t numOfCols) {
  // | total rows/total length | block group id | column schema | each column length |
L
Liu Jicong 已提交
687 688
  return sizeof(int32_t) + sizeof(uint64_t) + numOfCols * (sizeof(int16_t) + sizeof(int32_t)) +
         numOfCols * sizeof(int32_t);
H
Haojun Liao 已提交
689 690
}

X
Xiaoyu Wang 已提交
691
double blockDataGetSerialRowSize(const SSDataBlock* pBlock) {
H
Haojun Liao 已提交
692 693 694 695
  ASSERT(pBlock != NULL);
  double rowSize = 0;

  size_t numOfCols = pBlock->info.numOfCols;
L
Liu Jicong 已提交
696
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
697 698 699 700 701 702
    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
    rowSize += pColInfo->info.bytes;

    if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {
      rowSize += sizeof(int32_t);
    } else {
L
Liu Jicong 已提交
703
      rowSize += 1 / 8.0;  // one bit for each record
H
Haojun Liao 已提交
704 705 706 707 708 709
    }
  }

  return rowSize;
}

H
Haojun Liao 已提交
710
typedef struct SSDataBlockSortHelper {
L
Liu Jicong 已提交
711 712
  SArray*      orderInfo;  // SArray<SBlockOrderInfo>
  SSDataBlock* pDataBlock;
H
Haojun Liao 已提交
713 714 715
} SSDataBlockSortHelper;

int32_t dataBlockCompar(const void* p1, const void* p2, const void* param) {
L
Liu Jicong 已提交
716
  const SSDataBlockSortHelper* pHelper = (const SSDataBlockSortHelper*)param;
H
Haojun Liao 已提交
717 718 719

  SSDataBlock* pDataBlock = pHelper->pDataBlock;

L
Liu Jicong 已提交
720 721
  int32_t left = *(int32_t*)p1;
  int32_t right = *(int32_t*)p2;
H
Haojun Liao 已提交
722 723

  SArray* pInfo = pHelper->orderInfo;
724

L
Liu Jicong 已提交
725
  for (int32_t i = 0; i < pInfo->size; ++i) {
H
Haojun Liao 已提交
726
    SBlockOrderInfo* pOrder = TARRAY_GET_ELEM(pInfo, i);
L
Liu Jicong 已提交
727
    SColumnInfoData* pColInfoData = pOrder->pColData;  // TARRAY_GET_ELEM(pDataBlock->pDataBlock, pOrder->colIndex);
H
Haojun Liao 已提交
728

729
    if (pColInfoData->hasNull) {
730 731
      bool leftNull = colDataIsNull(pColInfoData, pDataBlock->info.rows, left, NULL);
      bool rightNull = colDataIsNull(pColInfoData, pDataBlock->info.rows, right, NULL);
732
      if (leftNull && rightNull) {
L
Liu Jicong 已提交
733
        continue;  // continue to next slot
734
      }
H
Haojun Liao 已提交
735

736
      if (rightNull) {
H
Haojun Liao 已提交
737
        return pOrder->nullFirst ? 1 : -1;
738
      }
H
Haojun Liao 已提交
739

740
      if (leftNull) {
H
Haojun Liao 已提交
741
        return pOrder->nullFirst ? -1 : 1;
742
      }
H
Haojun Liao 已提交
743 744
    }

L
Liu Jicong 已提交
745
    void* left1 = colDataGetData(pColInfoData, left);
746
    void* right1 = colDataGetData(pColInfoData, right);
H
Haojun Liao 已提交
747

748
    __compar_fn_t fn = getKeyComparFunc(pColInfoData->info.type, pOrder->order);
749

750 751 752 753 754
    int ret = fn(left1, right1);
    if (ret == 0) {
      continue;
    } else {
      return ret;
H
Haojun Liao 已提交
755 756 757 758 759 760
    }
  }

  return 0;
}

L
Liu Jicong 已提交
761 762
static int32_t doAssignOneTuple(SColumnInfoData* pDstCols, int32_t numOfRows, const SSDataBlock* pSrcBlock,
                                int32_t tupleIndex) {
H
Haojun Liao 已提交
763
  int32_t code = 0;
H
Haojun Liao 已提交
764 765 766 767 768 769
  int32_t numOfCols = pSrcBlock->info.numOfCols;

  for (int32_t i = 0; i < numOfCols; ++i) {
    SColumnInfoData* pDst = &pDstCols[i];
    SColumnInfoData* pSrc = taosArrayGet(pSrcBlock->pDataBlock, i);

770
    if (pSrc->hasNull && colDataIsNull(pSrc, pSrcBlock->info.rows, tupleIndex, pSrcBlock->pBlockAgg[i])) {
H
Haojun Liao 已提交
771 772 773 774
      code = colDataAppend(pDst, numOfRows, NULL, true);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
H
Haojun Liao 已提交
775
    } else {
776
      char* p = colDataGetData(pSrc, tupleIndex);
H
Haojun Liao 已提交
777 778 779 780
      code = colDataAppend(pDst, numOfRows, p, false);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
H
Haojun Liao 已提交
781 782
    }
  }
783 784

  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
785 786
}

H
Haojun Liao 已提交
787
static int32_t blockDataAssign(SColumnInfoData* pCols, const SSDataBlock* pDataBlock, int32_t* index) {
H
Haojun Liao 已提交
788
#if 0
H
Haojun Liao 已提交
789
  for (int32_t i = 0; i < pDataBlock->info.rows; ++i) {
H
Haojun Liao 已提交
790 791 792 793
    int32_t code = doAssignOneTuple(pCols, i, pDataBlock, index[i]);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
H
Haojun Liao 已提交
794
  }
H
Haojun Liao 已提交
795
#else
L
Liu Jicong 已提交
796
  for (int32_t i = 0; i < pDataBlock->info.numOfCols; ++i) {
H
Haojun Liao 已提交
797 798 799 800
    SColumnInfoData* pDst = &pCols[i];
    SColumnInfoData* pSrc = taosArrayGet(pDataBlock->pDataBlock, i);

    if (IS_VAR_DATA_TYPE(pSrc->info.type)) {
L
Liu Jicong 已提交
801 802
      memcpy(pDst->pData, pSrc->pData, pSrc->varmeta.length);
      pDst->varmeta.length = pSrc->varmeta.length;
H
Haojun Liao 已提交
803

L
Liu Jicong 已提交
804 805 806
      for (int32_t j = 0; j < pDataBlock->info.rows; ++j) {
        pDst->varmeta.offset[j] = pSrc->varmeta.offset[index[j]];
      }
H
Haojun Liao 已提交
807
    } else {
wmmhello's avatar
wmmhello 已提交
808 809 810 811
      for (int32_t j = 0; j < pDataBlock->info.rows; ++j) {
        if (colDataIsNull_f(pSrc->nullbitmap, index[j])) {
          colDataSetNull_f(pDst->nullbitmap, j);
          continue;
H
Haojun Liao 已提交
812
        }
wmmhello's avatar
wmmhello 已提交
813
        memcpy(pDst->pData + j * pDst->info.bytes, pSrc->pData + index[j] * pDst->info.bytes, pDst->info.bytes);
H
Haojun Liao 已提交
814 815 816 817
      }
    }
  }
#endif
H
Haojun Liao 已提交
818
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
819 820 821 822 823 824
}

static SColumnInfoData* createHelpColInfoData(const SSDataBlock* pDataBlock) {
  int32_t rows = pDataBlock->info.rows;
  int32_t numOfCols = pDataBlock->info.numOfCols;

wafwerar's avatar
wafwerar 已提交
825
  SColumnInfoData* pCols = taosMemoryCalloc(numOfCols, sizeof(SColumnInfoData));
H
Haojun Liao 已提交
826 827 828 829
  if (pCols == NULL) {
    return NULL;
  }

L
Liu Jicong 已提交
830
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
831 832 833 834
    SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, i);
    pCols[i].info = pColInfoData->info;

    if (IS_VAR_DATA_TYPE(pCols[i].info.type)) {
wafwerar's avatar
wafwerar 已提交
835 836
      pCols[i].varmeta.offset = taosMemoryCalloc(rows, sizeof(int32_t));
      pCols[i].pData = taosMemoryCalloc(1, pColInfoData->varmeta.length);
H
Haojun Liao 已提交
837 838 839

      pCols[i].varmeta.length = pColInfoData->varmeta.length;
      pCols[i].varmeta.allocLen = pCols[i].varmeta.length;
H
Haojun Liao 已提交
840
    } else {
wafwerar's avatar
wafwerar 已提交
841 842
      pCols[i].nullbitmap = taosMemoryCalloc(1, BitmapLen(rows));
      pCols[i].pData = taosMemoryCalloc(rows, pCols[i].info.bytes);
H
Haojun Liao 已提交
843 844 845 846 847 848
    }
  }

  return pCols;
}

H
Haojun Liao 已提交
849
static void copyBackToBlock(SSDataBlock* pDataBlock, SColumnInfoData* pCols) {
H
Haojun Liao 已提交
850 851
  int32_t numOfCols = pDataBlock->info.numOfCols;

L
Liu Jicong 已提交
852
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
853 854 855 856
    SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, i);
    pColInfoData->info = pCols[i].info;

    if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
wafwerar's avatar
wafwerar 已提交
857
      taosMemoryFreeClear(pColInfoData->varmeta.offset);
H
Haojun Liao 已提交
858 859
      pColInfoData->varmeta = pCols[i].varmeta;
    } else {
wafwerar's avatar
wafwerar 已提交
860
      taosMemoryFreeClear(pColInfoData->nullbitmap);
H
Haojun Liao 已提交
861 862 863
      pColInfoData->nullbitmap = pCols[i].nullbitmap;
    }

wafwerar's avatar
wafwerar 已提交
864
    taosMemoryFreeClear(pColInfoData->pData);
H
Haojun Liao 已提交
865 866 867
    pColInfoData->pData = pCols[i].pData;
  }

wafwerar's avatar
wafwerar 已提交
868
  taosMemoryFreeClear(pCols);
H
Haojun Liao 已提交
869 870 871
}

static int32_t* createTupleIndex(size_t rows) {
wafwerar's avatar
wafwerar 已提交
872
  int32_t* index = taosMemoryCalloc(rows, sizeof(int32_t));
H
Haojun Liao 已提交
873 874 875 876
  if (index == NULL) {
    return NULL;
  }

L
Liu Jicong 已提交
877
  for (int32_t i = 0; i < rows; ++i) {
H
Haojun Liao 已提交
878 879 880 881 882 883
    index[i] = i;
  }

  return index;
}

wafwerar's avatar
wafwerar 已提交
884
static void destroyTupleIndex(int32_t* index) { taosMemoryFreeClear(index); }
H
Haojun Liao 已提交
885

H
Haojun Liao 已提交
886
int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo) {
H
Haojun Liao 已提交
887 888 889 890 891 892 893
  ASSERT(pDataBlock != NULL && pOrderInfo != NULL);
  if (pDataBlock->info.rows <= 1) {
    return TSDB_CODE_SUCCESS;
  }

  // Allocate the additional buffer.
  uint32_t rows = pDataBlock->info.rows;
H
Haojun Liao 已提交
894 895 896 897 898 899 900

  bool sortColumnHasNull = false;
  bool varTypeSort = false;

  for (int32_t i = 0; i < taosArrayGetSize(pOrderInfo); ++i) {
    SBlockOrderInfo* pInfo = taosArrayGet(pOrderInfo, i);

H
Haojun Liao 已提交
901
    SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, pInfo->slotId);
H
Haojun Liao 已提交
902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918
    if (pColInfoData->hasNull) {
      sortColumnHasNull = true;
    }

    if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
      varTypeSort = true;
    }
  }

  if (taosArrayGetSize(pOrderInfo) == 1 && (!sortColumnHasNull)) {
    if (pDataBlock->info.numOfCols == 1) {
      if (!varTypeSort) {
        SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, 0);
        SBlockOrderInfo* pOrder = taosArrayGet(pOrderInfo, 0);

        int64_t p0 = taosGetTimestampUs();

919
        __compar_fn_t fn = getKeyComparFunc(pColInfoData->info.type, pOrder->order);
H
Haojun Liao 已提交
920 921 922
        qsort(pColInfoData->pData, pDataBlock->info.rows, pColInfoData->info.bytes, fn);

        int64_t p1 = taosGetTimestampUs();
923
        uDebug("blockDataSort easy cost:%" PRId64 ", rows:%d\n", p1 - p0, pDataBlock->info.rows);
H
Haojun Liao 已提交
924 925 926 927 928 929 930 931

        return TSDB_CODE_SUCCESS;
      } else {  // var data type
      }
    } else if (pDataBlock->info.numOfCols == 2) {
    }
  }

H
Haojun Liao 已提交
932 933 934 935 936 937
  int32_t* index = createTupleIndex(rows);
  if (index == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return terrno;
  }

938 939
  int64_t p0 = taosGetTimestampUs();

H
Haojun Liao 已提交
940
  SSDataBlockSortHelper helper = {.pDataBlock = pDataBlock, .orderInfo = pOrderInfo};
L
Liu Jicong 已提交
941
  for (int32_t i = 0; i < taosArrayGetSize(helper.orderInfo); ++i) {
H
Haojun Liao 已提交
942
    struct SBlockOrderInfo* pInfo = taosArrayGet(helper.orderInfo, i);
H
Haojun Liao 已提交
943
    pInfo->pColData = taosArrayGet(pDataBlock->pDataBlock, pInfo->slotId);
H
Haojun Liao 已提交
944 945
  }

H
Haojun Liao 已提交
946 947
  taosqsort(index, rows, sizeof(int32_t), &helper, dataBlockCompar);

948 949
  int64_t p1 = taosGetTimestampUs();

H
Haojun Liao 已提交
950 951
  SColumnInfoData* pCols = createHelpColInfoData(pDataBlock);
  if (pCols == NULL) {
952
    destroyTupleIndex(index);
H
Haojun Liao 已提交
953 954 955 956
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return terrno;
  }

957 958
  int64_t p2 = taosGetTimestampUs();

wmmhello's avatar
wmmhello 已提交
959
  blockDataAssign(pCols, pDataBlock, index);
H
Haojun Liao 已提交
960

961
  int64_t p3 = taosGetTimestampUs();
H
Haojun Liao 已提交
962 963

  copyBackToBlock(pDataBlock, pCols);
964 965
  int64_t p4 = taosGetTimestampUs();

C
Cary Xu 已提交
966 967 968
  uDebug("blockDataSort complex sort:%" PRId64 ", create:%" PRId64 ", assign:%" PRId64 ", copyback:%" PRId64
         ", rows:%d\n",
         p1 - p0, p2 - p1, p3 - p2, p4 - p3, rows);
H
Haojun Liao 已提交
969
  destroyTupleIndex(index);
H
Haojun Liao 已提交
970 971

  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
972
}
973

H
Haojun Liao 已提交
974 975
typedef struct SHelper {
  int32_t index;
L
Liu Jicong 已提交
976 977 978 979 980
  union {
    char*   pData;
    int64_t i64;
    double  d64;
  };
H
Haojun Liao 已提交
981 982 983 984 985 986
} SHelper;

SHelper* createTupleIndex_rv(int32_t numOfRows, SArray* pOrderInfo, SSDataBlock* pBlock) {
  int32_t sortValLengthPerRow = 0;
  int32_t numOfCols = taosArrayGetSize(pOrderInfo);

L
Liu Jicong 已提交
987
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
988
    SBlockOrderInfo* pInfo = taosArrayGet(pOrderInfo, i);
H
Haojun Liao 已提交
989
    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, pInfo->slotId);
H
Haojun Liao 已提交
990 991 992 993 994 995
    pInfo->pColData = pColInfo;
    sortValLengthPerRow += pColInfo->info.bytes;
  }

  size_t len = sortValLengthPerRow * pBlock->info.rows;

wafwerar's avatar
wafwerar 已提交
996 997
  char*    buf = taosMemoryCalloc(1, len);
  SHelper* phelper = taosMemoryCalloc(numOfRows, sizeof(SHelper));
L
Liu Jicong 已提交
998
  for (int32_t i = 0; i < numOfRows; ++i) {
H
Haojun Liao 已提交
999 1000 1001 1002 1003
    phelper[i].index = i;
    phelper[i].pData = buf + sortValLengthPerRow * i;
  }

  int32_t offset = 0;
L
Liu Jicong 已提交
1004
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
1005
    SBlockOrderInfo* pInfo = taosArrayGet(pOrderInfo, i);
L
Liu Jicong 已提交
1006 1007 1008 1009
    for (int32_t j = 0; j < numOfRows; ++j) {
      phelper[j].i64 = *(int32_t*)pInfo->pColData->pData + pInfo->pColData->info.bytes * j;
      //      memcpy(phelper[j].pData + offset, pInfo->pColData->pData + pInfo->pColData->info.bytes * j,
      //      pInfo->pColData->info.bytes);
H
Haojun Liao 已提交
1010 1011 1012 1013 1014 1015 1016 1017 1018
    }

    offset += pInfo->pColData->info.bytes;
  }

  return phelper;
}

int32_t dataBlockCompar_rv(const void* p1, const void* p2, const void* param) {
L
Liu Jicong 已提交
1019
  const SSDataBlockSortHelper* pHelper = (const SSDataBlockSortHelper*)param;
H
Haojun Liao 已提交
1020

L
Liu Jicong 已提交
1021
  //  SSDataBlock* pDataBlock = pHelper->pDataBlock;
H
Haojun Liao 已提交
1022

L
Liu Jicong 已提交
1023 1024
  SHelper* left = (SHelper*)p1;
  SHelper* right = (SHelper*)p2;
H
Haojun Liao 已提交
1025 1026 1027 1028

  SArray* pInfo = pHelper->orderInfo;

  int32_t offset = 0;
L
Liu Jicong 已提交
1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072
  //  for(int32_t i = 0; i < pInfo->size; ++i) {
  //    SBlockOrderInfo* pOrder = TARRAY_GET_ELEM(pInfo, 0);
  //    SColumnInfoData* pColInfoData = pOrder->pColData;//TARRAY_GET_ELEM(pDataBlock->pDataBlock, pOrder->colIndex);

  //    if (pColInfoData->hasNull) {
  //      bool leftNull  = colDataIsNull(pColInfoData, pDataBlock->info.rows, left, pDataBlock->pBlockAgg);
  //      bool rightNull = colDataIsNull(pColInfoData, pDataBlock->info.rows, right, pDataBlock->pBlockAgg);
  //      if (leftNull && rightNull) {
  //        continue; // continue to next slot
  //      }
  //
  //      if (rightNull) {
  //        return pHelper->nullFirst? 1:-1;
  //      }
  //
  //      if (leftNull) {
  //        return pHelper->nullFirst? -1:1;
  //      }
  //    }

  //    void* left1  = colDataGetData(pColInfoData, left);
  //    void* right1 = colDataGetData(pColInfoData, right);

  //    switch(pColInfoData->info.type) {
  //      case TSDB_DATA_TYPE_INT: {
  int32_t leftx = *(int32_t*)left->pData;    //*(int32_t*)(left->pData + offset);
  int32_t rightx = *(int32_t*)right->pData;  //*(int32_t*)(right->pData + offset);

  //        offset += pColInfoData->info.bytes;
  if (leftx == rightx) {
    //          break;
    return 0;
  } else {
    //          if (pOrder->order == TSDB_ORDER_ASC) {
    return (leftx < rightx) ? -1 : 1;
    //          } else {
    //            return (leftx < rightx)? 1:-1;
    //          }
  }
  //      }
  //      default:
  //        assert(0);
  //    }
  //  }
H
Haojun Liao 已提交
1073 1074 1075 1076

  return 0;
}

L
Liu Jicong 已提交
1077
int32_t varColSort(SColumnInfoData* pColumnInfoData, SBlockOrderInfo* pOrder) { return 0; }
H
Haojun Liao 已提交
1078 1079

int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst) {
L
Liu Jicong 已提交
1080
  // Allocate the additional buffer.
H
Haojun Liao 已提交
1081 1082
  int64_t p0 = taosGetTimestampUs();

H
Haojun Liao 已提交
1083
  SSDataBlockSortHelper helper = {.pDataBlock = pDataBlock, .orderInfo = pOrderInfo};
H
Haojun Liao 已提交
1084 1085 1086 1087 1088 1089 1090 1091 1092 1093

  uint32_t rows = pDataBlock->info.rows;
  SHelper* index = createTupleIndex_rv(rows, helper.orderInfo, pDataBlock);
  if (index == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return terrno;
  }

  taosqsort(index, rows, sizeof(SHelper), &helper, dataBlockCompar_rv);

L
Liu Jicong 已提交
1094
  int64_t          p1 = taosGetTimestampUs();
H
Haojun Liao 已提交
1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113
  SColumnInfoData* pCols = createHelpColInfoData(pDataBlock);
  if (pCols == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return terrno;
  }

  int64_t p2 = taosGetTimestampUs();

  //  int32_t code = blockDataAssign(pCols, pDataBlock, index);
  //  if (code != TSDB_CODE_SUCCESS) {
  //    terrno = code;
  //    return code;
  //  }

  int64_t p3 = taosGetTimestampUs();

  copyBackToBlock(pDataBlock, pCols);
  int64_t p4 = taosGetTimestampUs();

L
Liu Jicong 已提交
1114 1115
  printf("sort:%" PRId64 ", create:%" PRId64 ", assign:%" PRId64 ", copyback:%" PRId64 ", rows:%d\n", p1 - p0, p2 - p1,
         p3 - p2, p4 - p3, rows);
H
Haojun Liao 已提交
1116
  //  destroyTupleIndex(index);
1117
  return 0;
H
Haojun Liao 已提交
1118 1119
}

1120
void blockDataCleanup(SSDataBlock* pDataBlock) {
1121
  pDataBlock->info.rows = 0;
1122 1123 1124
  for (int32_t i = 0; i < pDataBlock->info.numOfCols; ++i) {
    SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
    colInfoDataCleanup(p, pDataBlock->info.capacity);
1125 1126 1127
  }
}

1128 1129
int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, size_t existRows, uint32_t numOfRows) {
  if (0 == numOfRows || numOfRows <= existRows) {
X
Xiaoyu Wang 已提交
1130 1131 1132
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
1133
  if (IS_VAR_DATA_TYPE(pColumn->info.type)) {
wafwerar's avatar
wafwerar 已提交
1134
    char* tmp = taosMemoryRealloc(pColumn->varmeta.offset, sizeof(int32_t) * numOfRows);
1135

D
dapan1121 已提交
1136 1137 1138
    if (tmp == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
1139

D
dapan1121 已提交
1140
    pColumn->varmeta.offset = (int32_t*)tmp;
1141
    memset(&pColumn->varmeta.offset[existRows], 0, sizeof(int32_t) * (numOfRows - existRows));
D
dapan1121 已提交
1142
  } else {
wafwerar's avatar
wafwerar 已提交
1143
    char* tmp = taosMemoryRealloc(pColumn->nullbitmap, BitmapLen(numOfRows));
D
dapan1121 已提交
1144 1145 1146
    if (tmp == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
1147

1148
    int32_t oldLen = BitmapLen(existRows);
D
dapan1121 已提交
1149
    pColumn->nullbitmap = tmp;
1150
    memset(&pColumn->nullbitmap[oldLen], 0, BitmapLen(numOfRows) - oldLen);
1151 1152 1153 1154 1155

    if (pColumn->info.type == TSDB_DATA_TYPE_NULL) {
      return TSDB_CODE_SUCCESS;
    }

X
Xiaoyu Wang 已提交
1156
    assert(pColumn->info.bytes);
wafwerar's avatar
wafwerar 已提交
1157
    tmp = taosMemoryRealloc(pColumn->pData, numOfRows * pColumn->info.bytes);
D
dapan1121 已提交
1158 1159 1160
    if (tmp == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
1161
    memset(tmp + pColumn->info.bytes * existRows, 0, pColumn->info.bytes * (numOfRows - existRows));
1162

D
dapan1121 已提交
1163 1164 1165 1166 1167 1168
    pColumn->pData = tmp;
  }

  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
1169
void colInfoDataCleanup(SColumnInfoData* pColumn, uint32_t numOfRows) {
1170 1171 1172
  if (IS_VAR_DATA_TYPE(pColumn->info.type)) {
    pColumn->varmeta.length = 0;
  } else {
dengyihao's avatar
dengyihao 已提交
1173 1174 1175
    if (pColumn->nullbitmap != NULL) {
      memset(pColumn->nullbitmap, 0, BitmapLen(numOfRows));
    }
1176 1177 1178
  }
}

D
dapan1121 已提交
1179 1180
int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows) {
  int32_t code = 0;
H
Haojun Liao 已提交
1181 1182 1183
  if (numOfRows == 0) {
    return TSDB_CODE_SUCCESS;
  }
L
Liu Jicong 已提交
1184

1185 1186
  pDataBlock->info.capacity = numOfRows;

L
Liu Jicong 已提交
1187
  for (int32_t i = 0; i < pDataBlock->info.numOfCols; ++i) {
D
dapan1121 已提交
1188
    SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
1189
    code = colInfoDataEnsureCapacity(p, pDataBlock->info.rows, numOfRows);
D
dapan1121 已提交
1190 1191
    if (code) {
      return code;
1192 1193
    }
  }
H
Haojun Liao 已提交
1194 1195 1196 1197 1198 1199 1200 1201 1202

  return TSDB_CODE_SUCCESS;
}

void* blockDataDestroy(SSDataBlock* pBlock) {
  if (pBlock == NULL) {
    return NULL;
  }

L
Liu Jicong 已提交
1203
  blockDestroyInner(pBlock);
wafwerar's avatar
wafwerar 已提交
1204
  taosMemoryFreeClear(pBlock);
H
Haojun Liao 已提交
1205
  return NULL;
D
dapan1121 已提交
1206
}
1207

1208
SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData) {
C
Cary Xu 已提交
1209
  if (pDataBlock == NULL) {
1210 1211
    return NULL;
  }
1212

1213 1214
  int32_t numOfCols = pDataBlock->info.numOfCols;

wafwerar's avatar
wafwerar 已提交
1215
  SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
1216
  pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
1217

1218
  pBlock->info.numOfCols = numOfCols;
1219
  pBlock->info.hasVarCol = pDataBlock->info.hasVarCol;
1220
  pBlock->info.rowSize = pDataBlock->info.rowSize;
1221
  pBlock->info.groupId = pDataBlock->info.groupId;
1222 1223
  pBlock->info.childId = pDataBlock->info.childId;
  pBlock->info.type = pDataBlock->info.type;
1224

L
Liu Jicong 已提交
1225
  for (int32_t i = 0; i < numOfCols; ++i) {
1226 1227 1228
    SColumnInfoData  colInfo = {0};
    SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
    colInfo.info = p->info;
1229
    colInfo.hasNull = true;
1230 1231 1232
    taosArrayPush(pBlock->pDataBlock, &colInfo);
  }

1233 1234 1235 1236 1237
  if (copyData) {
    for (int32_t i = 0; i < numOfCols; ++i) {
      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, i);
      SColumnInfoData* pSrc = taosArrayGet(pDataBlock->pDataBlock, i);

1238
      int32_t code = colInfoDataEnsureCapacity(pDst, 0, pDataBlock->info.rows);
1239 1240 1241 1242
      if (code != TSDB_CODE_SUCCESS) {
        return NULL;
      }

1243
      if (pSrc->pData == NULL) {
1244 1245
        continue;
      }
1246 1247 1248 1249 1250 1251 1252
      colDataAssign(pDst, pSrc, pDataBlock->info.rows);
    }

    pBlock->info.rows = pDataBlock->info.rows;
    pBlock->info.capacity = pDataBlock->info.rows;
  }

1253 1254 1255 1256
  return pBlock;
}

size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize) {
1257
  int32_t payloadSize = pageSize - blockDataGetSerialMetaSize(pBlock->info.numOfCols);
1258 1259 1260 1261 1262 1263 1264

  int32_t rowSize = pBlock->info.rowSize;

  int32_t nRows = payloadSize / rowSize;

  // the true value must be less than the value of nRows
  int32_t additional = 0;
X
Xiaoyu Wang 已提交
1265
  for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
1266 1267 1268 1269 1270 1271 1272 1273 1274
    SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);
    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
      additional += nRows * sizeof(int32_t);
    } else {
      additional += BitmapLen(nRows);
    }
  }

  int32_t newRows = (payloadSize - additional) / rowSize;
1275
  ASSERT(newRows <= nRows && newRows >= 1);
1276 1277

  return newRows;
1278
}
H
Haojun Liao 已提交
1279

1280 1281
void colDataDestroy(SColumnInfoData* pColData) {
  if (IS_VAR_DATA_TYPE(pColData->info.type)) {
wmmhello's avatar
wmmhello 已提交
1282
    taosMemoryFreeClear(pColData->varmeta.offset);
1283
  } else {
wmmhello's avatar
wmmhello 已提交
1284
    taosMemoryFreeClear(pColData->nullbitmap);
1285 1286
  }

wmmhello's avatar
wmmhello 已提交
1287
  taosMemoryFreeClear(pColData->pData);
1288 1289
}

H
Haojun Liao 已提交
1290 1291 1292 1293
static void doShiftBitmap(char* nullBitmap, size_t n, size_t total) {
  int32_t len = BitmapLen(total);

  int32_t newLen = BitmapLen(total - n);
C
Cary Xu 已提交
1294 1295
  if (n % 8 == 0) {
    memmove(nullBitmap, nullBitmap + n / 8, newLen);
H
Haojun Liao 已提交
1296
  } else {
1297
    int32_t  tail = n % 8;
1298
    int32_t  i = 0;
C
Cary Xu 已提交
1299
    uint8_t* p = (uint8_t*)nullBitmap;
H
Haojun Liao 已提交
1300

1301 1302 1303 1304 1305 1306 1307 1308 1309
    if (n < 8) {
      while (i < len) {
        uint8_t v = p[i];  // source bitmap value
        p[i] = (v << tail);

        if (i < len - 1) {
          uint8_t next = p[i + 1];
          p[i] |= (next >> (8 - tail));
        }
H
Haojun Liao 已提交
1310

1311
        i += 1;
H
Haojun Liao 已提交
1312
      }
1313 1314
    } else if (n > 8) {
      int32_t gap = len - newLen;
1315
      while (i < newLen) {
1316 1317 1318 1319 1320 1321 1322
        uint8_t v = p[i + gap];
        p[i] = (v << tail);

        if (i < newLen - 1) {
          uint8_t next = p[i + gap + 1];
          p[i] |= (next >> (8 - tail));
        }
H
Haojun Liao 已提交
1323

1324 1325
        i += 1;
      }
H
Haojun Liao 已提交
1326 1327 1328 1329 1330 1331
    }
  }
}

static void colDataTrimFirstNRows(SColumnInfoData* pColInfoData, size_t n, size_t total) {
  if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
1332 1333
    memmove(pColInfoData->varmeta.offset, &pColInfoData->varmeta.offset[n], (total - n) * sizeof(int32_t));
    memset(&pColInfoData->varmeta.offset[total - n], 0, n);
H
Haojun Liao 已提交
1334 1335 1336 1337 1338 1339 1340
  } else {
    int32_t bytes = pColInfoData->info.bytes;
    memmove(pColInfoData->pData, ((char*)pColInfoData->pData + n * bytes), (total - n) * bytes);
    doShiftBitmap(pColInfoData->nullbitmap, n, total);
  }
}

C
Cary Xu 已提交
1341
int32_t blockDataTrimFirstNRows(SSDataBlock* pBlock, size_t n) {
H
Haojun Liao 已提交
1342 1343 1344 1345 1346 1347 1348
  if (n == 0) {
    return TSDB_CODE_SUCCESS;
  }

  if (pBlock->info.rows <= n) {
    blockDataCleanup(pBlock);
  } else {
C
Cary Xu 已提交
1349
    for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
H
Haojun Liao 已提交
1350 1351 1352 1353 1354 1355 1356 1357 1358
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
      colDataTrimFirstNRows(pColInfoData, n, pBlock->info.rows);
    }

    pBlock->info.rows -= n;
  }
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376
int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
  int64_t tbUid = pBlock->info.uid;
  int16_t numOfCols = pBlock->info.numOfCols;
  int16_t hasVarCol = pBlock->info.hasVarCol;
  int32_t rows = pBlock->info.rows;
  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);

  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, tbUid);
  tlen += taosEncodeFixedI16(buf, numOfCols);
  tlen += taosEncodeFixedI16(buf, hasVarCol);
  tlen += taosEncodeFixedI32(buf, rows);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SColumnInfoData* pColData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
    tlen += taosEncodeFixedI16(buf, pColData->info.colId);
    tlen += taosEncodeFixedI16(buf, pColData->info.type);
    tlen += taosEncodeFixedI32(buf, pColData->info.bytes);
S
slzhou@taodata.com 已提交
1377
    tlen += taosEncodeFixedBool(buf, pColData->hasNull);
H
Haojun Liao 已提交
1378 1379 1380 1381 1382 1383 1384 1385

    if (IS_VAR_DATA_TYPE(pColData->info.type)) {
      tlen += taosEncodeBinary(buf, pColData->varmeta.offset, sizeof(int32_t) * rows);
    } else {
      tlen += taosEncodeBinary(buf, pColData->nullbitmap, BitmapLen(rows));
    }

    int32_t len = colDataGetLength(pColData, rows);
X
Xiaoyu Wang 已提交
1386
    tlen += taosEncodeFixedI32(buf, len);
H
Haojun Liao 已提交
1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406

    tlen += taosEncodeBinary(buf, pColData->pData, len);
  }
  return tlen;
}

void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock) {
  int32_t sz;

  buf = taosDecodeFixedI64(buf, &pBlock->info.uid);
  buf = taosDecodeFixedI16(buf, &pBlock->info.numOfCols);
  buf = taosDecodeFixedI16(buf, &pBlock->info.hasVarCol);
  buf = taosDecodeFixedI32(buf, &pBlock->info.rows);
  buf = taosDecodeFixedI32(buf, &sz);
  pBlock->pDataBlock = taosArrayInit(sz, sizeof(SColumnInfoData));
  for (int32_t i = 0; i < sz; i++) {
    SColumnInfoData data = {0};
    buf = taosDecodeFixedI16(buf, &data.info.colId);
    buf = taosDecodeFixedI16(buf, &data.info.type);
    buf = taosDecodeFixedI32(buf, &data.info.bytes);
S
slzhou@taodata.com 已提交
1407
    buf = taosDecodeFixedBool(buf, &data.hasNull);
H
Haojun Liao 已提交
1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422

    if (IS_VAR_DATA_TYPE(data.info.type)) {
      buf = taosDecodeBinary(buf, (void**)&data.varmeta.offset, pBlock->info.rows * sizeof(int32_t));
      data.varmeta.length = pBlock->info.rows * sizeof(int32_t);
      data.varmeta.allocLen = data.varmeta.length;
    } else {
      buf = taosDecodeBinary(buf, (void**)&data.nullbitmap, BitmapLen(pBlock->info.rows));
    }

    int32_t len = 0;
    buf = taosDecodeFixedI32(buf, &len);
    buf = taosDecodeBinary(buf, (void**)&data.pData, len);
    taosArrayPush(pBlock->pDataBlock, &data);
  }
  return (void*)buf;
L
Liu Jicong 已提交
1423
}
L
Liu Jicong 已提交
1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437

int32_t tEncodeDataBlocks(void** buf, const SArray* blocks) {
  int32_t tlen = 0;
  int32_t sz = taosArrayGetSize(blocks);
  tlen += taosEncodeFixedI32(buf, sz);

  for (int32_t i = 0; i < sz; i++) {
    SSDataBlock* pBlock = taosArrayGet(blocks, i);
    tlen += tEncodeDataBlock(buf, pBlock);
  }

  return tlen;
}

L
Liu Jicong 已提交
1438
void* tDecodeDataBlocks(const void* buf, SArray** blocks) {
L
Liu Jicong 已提交
1439 1440
  int32_t sz;
  buf = taosDecodeFixedI32(buf, &sz);
L
Liu Jicong 已提交
1441 1442

  *blocks = taosArrayInit(sz, sizeof(SSDataBlock));
L
Liu Jicong 已提交
1443 1444 1445
  for (int32_t i = 0; i < sz; i++) {
    SSDataBlock pBlock = {0};
    buf = tDecodeDataBlock(buf, &pBlock);
L
Liu Jicong 已提交
1446
    taosArrayPush(*blocks, &pBlock);
L
Liu Jicong 已提交
1447 1448 1449
  }
  return (void*)buf;
}
L
Liu Jicong 已提交
1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486

static char* formatTimestamp(char* buf, int64_t val, int precision) {
  time_t  tt;
  int32_t ms = 0;
  if (precision == TSDB_TIME_PRECISION_NANO) {
    tt = (time_t)(val / 1000000000);
    ms = val % 1000000000;
  } else if (precision == TSDB_TIME_PRECISION_MICRO) {
    tt = (time_t)(val / 1000000);
    ms = val % 1000000;
  } else {
    tt = (time_t)(val / 1000);
    ms = val % 1000;
  }

  /* comment out as it make testcases like select_with_tags.sim fail.
    but in windows, this may cause the call to localtime crash if tt < 0,
    need to find a better solution.
    if (tt < 0) {
      tt = 0;
    }
    */

#ifdef WINDOWS
  if (tt < 0) tt = 0;
#endif
  if (tt <= 0 && ms < 0) {
    tt--;
    if (precision == TSDB_TIME_PRECISION_NANO) {
      ms += 1000000000;
    } else if (precision == TSDB_TIME_PRECISION_MICRO) {
      ms += 1000000;
    } else {
      ms += 1000;
    }
  }

1487
  struct tm* ptm = taosLocalTime(&tt, NULL);
L
Liu Jicong 已提交
1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499
  size_t     pos = strftime(buf, 35, "%Y-%m-%d %H:%M:%S", ptm);

  if (precision == TSDB_TIME_PRECISION_NANO) {
    sprintf(buf + pos, ".%09d", ms);
  } else if (precision == TSDB_TIME_PRECISION_MICRO) {
    sprintf(buf + pos, ".%06d", ms);
  } else {
    sprintf(buf + pos, ".%03d", ms);
  }

  return buf;
}
H
Haojun Liao 已提交
1500

C
Cary Xu 已提交
1501
void blockDebugShowData(const SArray* dataBlocks, const char* flag) {
C
Cary Xu 已提交
1502
  char    pBuf[128] = {0};
L
Liu Jicong 已提交
1503 1504 1505 1506 1507
  int32_t sz = taosArrayGetSize(dataBlocks);
  for (int32_t i = 0; i < sz; i++) {
    SSDataBlock* pDataBlock = taosArrayGet(dataBlocks, i);
    int32_t      colNum = pDataBlock->info.numOfCols;
    int32_t      rows = pDataBlock->info.rows;
1508
    printf("%s |block type %d |child id %d|\n", flag, (int32_t)pDataBlock->info.type, pDataBlock->info.childId);
L
Liu Jicong 已提交
1509
    for (int32_t j = 0; j < rows; j++) {
C
Cary Xu 已提交
1510
      printf("%s |", flag);
L
Liu Jicong 已提交
1511 1512 1513
      for (int32_t k = 0; k < colNum; k++) {
        SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
        void*            var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
5
54liuyao 已提交
1514
        if (colDataIsNull(pColInfoData, rows, j, NULL)) {
5
54liuyao 已提交
1515 1516 1517
          printf(" %15s |", "NULL");
          continue;
        }
L
Liu Jicong 已提交
1518 1519 1520 1521 1522 1523 1524 1525
        switch (pColInfoData->info.type) {
          case TSDB_DATA_TYPE_TIMESTAMP:
            formatTimestamp(pBuf, *(uint64_t*)var, TSDB_TIME_PRECISION_MILLI);
            printf(" %25s |", pBuf);
            break;
          case TSDB_DATA_TYPE_INT:
            printf(" %15d |", *(int32_t*)var);
            break;
L
Liu Jicong 已提交
1526 1527 1528
          case TSDB_DATA_TYPE_UINT:
            printf(" %15u |", *(uint32_t*)var);
            break;
L
Liu Jicong 已提交
1529 1530 1531
          case TSDB_DATA_TYPE_BIGINT:
            printf(" %15ld |", *(int64_t*)var);
            break;
L
Liu Jicong 已提交
1532 1533 1534
          case TSDB_DATA_TYPE_UBIGINT:
            printf(" %15lu |", *(uint64_t*)var);
            break;
C
Cary Xu 已提交
1535 1536 1537
          case TSDB_DATA_TYPE_FLOAT:
            printf(" %15f |", *(float*)var);
            break;
5
54liuyao 已提交
1538
          case TSDB_DATA_TYPE_DOUBLE:
C
Cary Xu 已提交
1539
            printf(" %15lf |", *(double*)var);
5
54liuyao 已提交
1540
            break;
L
Liu Jicong 已提交
1541 1542 1543 1544 1545 1546 1547
        }
      }
      printf("\n");
    }
  }
}

C
Cary Xu 已提交
1548 1549 1550 1551 1552 1553 1554
/**
 * @brief TODO: Assume that the final generated result it less than 3M
 *
 * @param pReq
 * @param pDataBlocks
 * @param vgId
 * @param suid  // TODO: check with Liao whether suid response is reasonable
1555
 *
C
Cary Xu 已提交
1556 1557
 * TODO: colId should be set
 */
1558 1559
int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks, STSchema* pTSchema, int32_t vgId,
                                    tb_uid_t suid) {
C
Cary Xu 已提交
1560 1561 1562 1563 1564 1565 1566 1567 1568
  int32_t sz = taosArrayGetSize(pDataBlocks);
  int32_t bufSize = sizeof(SSubmitReq);
  for (int32_t i = 0; i < sz; ++i) {
    SDataBlockInfo* pBlkInfo = &((SSDataBlock*)taosArrayGet(pDataBlocks, i))->info;
    bufSize += pBlkInfo->rows * (TD_ROW_HEAD_LEN + pBlkInfo->rowSize + BitmapLen(pBlkInfo->numOfCols));
    bufSize += sizeof(SSubmitBlk);
  }

  *pReq = taosMemoryCalloc(1, bufSize);
1569
  if (!(*pReq)) {
C
Cary Xu 已提交
1570 1571 1572
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return TSDB_CODE_FAILED;
  }
C
Cary Xu 已提交
1573 1574
  void* pDataBuf = *pReq;

1575
  int32_t     msgLen = sizeof(SSubmitReq);
C
Cary Xu 已提交
1576 1577
  int32_t     numOfBlks = 0;
  SRowBuilder rb = {0};
C
Cary Xu 已提交
1578
  tdSRowInit(&rb, pTSchema->version);
C
Cary Xu 已提交
1579 1580 1581 1582 1583 1584 1585

  for (int32_t i = 0; i < sz; ++i) {
    SSDataBlock* pDataBlock = taosArrayGet(pDataBlocks, i);
    int32_t      colNum = pDataBlock->info.numOfCols;
    int32_t      rows = pDataBlock->info.rows;
    int32_t      rowSize = pDataBlock->info.rowSize;
    int64_t      groupId = pDataBlock->info.groupId;
1586 1587

    if (rb.nCols != colNum) {
C
Cary Xu 已提交
1588 1589 1590 1591 1592
      tdSRowSetTpInfo(&rb, colNum, pTSchema->flen);
    }

    SSubmitBlk* pSubmitBlk = POINTER_SHIFT(pDataBuf, msgLen);
    pSubmitBlk->suid = suid;
1593
    pSubmitBlk->uid = pDataBlock->info.groupId;
C
Cary Xu 已提交
1594 1595 1596 1597
    pSubmitBlk->numOfRows = rows;

    msgLen += sizeof(SSubmitBlk);
    int32_t dataLen = 0;
1598 1599
    for (int32_t j = 0; j < rows; ++j) {                     // iterate by row
      tdSRowResetBuf(&rb, POINTER_SHIFT(pDataBuf, msgLen));  // set row buf
1600
      bool    isStartKey = false;
1601
      int32_t offset = 0;
C
Cary Xu 已提交
1602 1603
      for (int32_t k = 0; k < colNum; ++k) {  // iterate by column
        SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
1604 1605
        STColumn*        pCol = &pTSchema->columns[k];
        void*            var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
C
Cary Xu 已提交
1606 1607 1608 1609
        switch (pColInfoData->info.type) {
          case TSDB_DATA_TYPE_TIMESTAMP:
            if (!isStartKey) {
              isStartKey = true;
1610
              tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NORM, var, true,
1611 1612
                                  offset, k);

C
Cary Xu 已提交
1613
            } else {
1614 1615
              tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NORM, var,
                                  true, offset, k);
C
Cary Xu 已提交
1616 1617 1618
            }
            break;
          case TSDB_DATA_TYPE_NCHAR: {
1619 1620
            tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_NCHAR, TD_VTYPE_NORM, var, true,
                                offset, k);
C
Cary Xu 已提交
1621 1622 1623
            break;
          }
          case TSDB_DATA_TYPE_VARCHAR: {  // TSDB_DATA_TYPE_BINARY
1624 1625
            tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_VARCHAR, TD_VTYPE_NORM, var, true,
                                offset, k);
C
Cary Xu 已提交
1626 1627 1628 1629 1630
            break;
          }
          case TSDB_DATA_TYPE_VARBINARY:
          case TSDB_DATA_TYPE_DECIMAL:
          case TSDB_DATA_TYPE_BLOB:
1631
          case TSDB_DATA_TYPE_JSON:
C
Cary Xu 已提交
1632
          case TSDB_DATA_TYPE_MEDIUMBLOB:
C
Cary Xu 已提交
1633
            uError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
C
Cary Xu 已提交
1634 1635 1636 1637
            TASSERT(0);
            break;
          default:
            if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
C
Cary Xu 已提交
1638 1639 1640
              if (pCol->type == pColInfoData->info.type) {
                tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, var, true, offset,
                                    k);
1641
              } else {
C
Cary Xu 已提交
1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661
                char tv[8] = {0};
                if (pColInfoData->info.type == TSDB_DATA_TYPE_FLOAT) {
                  float v = 0;
                  GET_TYPED_DATA(v, float, pColInfoData->info.type, var);
                  SET_TYPED_DATA(&tv, pCol->type, v);
                } else if (pColInfoData->info.type == TSDB_DATA_TYPE_DOUBLE) {
                  double v = 0;
                  GET_TYPED_DATA(v, double, pColInfoData->info.type, var);
                  SET_TYPED_DATA(&tv, pCol->type, v);
                } else if (IS_SIGNED_NUMERIC_TYPE(pColInfoData->info.type)) {
                  int64_t v = 0;
                  GET_TYPED_DATA(v, int64_t, pColInfoData->info.type, var);
                  SET_TYPED_DATA(&tv, pCol->type, v);
                } else {
                  uint64_t v = 0;
                  GET_TYPED_DATA(v, uint64_t, pColInfoData->info.type, var);
                  SET_TYPED_DATA(&tv, pCol->type, v);
                }
                tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, tv, true, offset,
                                    k);
1662
              }
C
Cary Xu 已提交
1663
            } else {
C
Cary Xu 已提交
1664
              uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
C
Cary Xu 已提交
1665 1666 1667 1668
              TASSERT(0);
            }
            break;
        }
1669
        offset += TYPE_BYTES[pCol->type];  // sum/avg would convert to int64_t/uint64_t/double during aggregation
C
Cary Xu 已提交
1670 1671
      }
      dataLen += TD_ROW_LEN(rb.pBuf);
C
Cary Xu 已提交
1672 1673 1674
#ifdef TD_DEBUG_PRINT_ROW
      tdSRowPrint(rb.pBuf, pTSchema, __func__);
#endif
C
Cary Xu 已提交
1675
    }
C
Cary Xu 已提交
1676 1677 1678

    ++numOfBlks;

C
Cary Xu 已提交
1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700
    pSubmitBlk->dataLen = dataLen;
    msgLen += pSubmitBlk->dataLen;
  }

  (*pReq)->length = msgLen;

  (*pReq)->header.vgId = htonl(vgId);
  (*pReq)->header.contLen = htonl(msgLen);
  (*pReq)->length = (*pReq)->header.contLen;
  (*pReq)->numOfBlocks = htonl(numOfBlks);
  SSubmitBlk* blk = (SSubmitBlk*)((*pReq) + 1);
  while (numOfBlks--) {
    int32_t dataLen = blk->dataLen;
    blk->uid = htobe64(blk->uid);
    blk->suid = htobe64(blk->suid);
    blk->padding = htonl(blk->padding);
    blk->sversion = htonl(blk->sversion);
    blk->dataLen = htonl(blk->dataLen);
    blk->schemaLen = htonl(blk->schemaLen);
    blk->numOfRows = htons(blk->numOfRows);
    blk = (SSubmitBlk*)(blk->data + dataLen);
  }
C
Cary Xu 已提交
1701 1702

  return TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
1703
}
L
Liu Jicong 已提交
1704

1705 1706 1707 1708 1709 1710 1711
char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId) {
  SArray* tags = taosArrayInit(0, sizeof(void*));
  SSmlKv* pTag = taosMemoryCalloc(1, sizeof(SSmlKv));
  pTag->key = "group_id";
  pTag->keyLen = strlen(pTag->key);
  pTag->type = TSDB_DATA_TYPE_UBIGINT;
  pTag->u = groupId;
L
Liu Jicong 已提交
1712
  pTag->length = sizeof(uint64_t);
1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732
  taosArrayPush(tags, &pTag);

  void* cname = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1);

  RandTableName rname = {
      .tags = tags,
      .sTableName = stbName,
      .sTableNameLen = strlen(stbName),
      .childTableName = cname,
  };

  buildChildTableName(&rname);

  taosMemoryFree(pTag);
  taosArrayDestroy(tags);

  ASSERT(rname.childTableName && rname.childTableName[0]);
  return rname.childTableName;
}

L
Liu Jicong 已提交
1733 1734
void blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols,
                         int8_t needCompress) {
1735
  // todo extract method
1736 1737 1738 1739 1740 1741
  int32_t* actualLen = (int32_t*)data;
  data += sizeof(int32_t);

  uint64_t* groupId = (uint64_t*)data;
  data += sizeof(uint64_t);

L
Liu Jicong 已提交
1742
  for (int32_t i = 0; i < numOfCols; ++i) {
1743
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
L
Liu Jicong 已提交
1744
    *((int16_t*)data) = pColInfoData->info.type;
1745 1746
    data += sizeof(int16_t);

L
Liu Jicong 已提交
1747
    *((int32_t*)data) = pColInfoData->info.bytes;
1748 1749 1750
    data += sizeof(int32_t);
  }

1751 1752 1753
  int32_t* colSizes = (int32_t*)data;
  data += numOfCols * sizeof(int32_t);

1754
  *dataLen = blockDataGetSerialMetaSize(numOfCols);
1755 1756 1757 1758 1759 1760

  int32_t numOfRows = pBlock->info.rows;
  for (int32_t col = 0; col < numOfCols; ++col) {
    SColumnInfoData* pColRes = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, col);

    // copy the null bitmap
1761
    size_t metaSize = 0;
1762
    if (IS_VAR_DATA_TYPE(pColRes->info.type)) {
1763
      metaSize = numOfRows * sizeof(int32_t);
1764 1765
      memcpy(data, pColRes->varmeta.offset, metaSize);
    } else {
1766 1767
      metaSize = BitmapLen(numOfRows);
      memcpy(data, pColRes->nullbitmap, metaSize);
1768 1769
    }

1770 1771 1772
    data += metaSize;
    (*dataLen) += metaSize;

1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792
    if (needCompress) {
      colSizes[col] = blockCompressColData(pColRes, numOfRows, data, needCompress);
      data += colSizes[col];
      (*dataLen) += colSizes[col];
    } else {
      colSizes[col] = colDataGetLength(pColRes, numOfRows);
      (*dataLen) += colSizes[col];
      memmove(data, pColRes->pData, colSizes[col]);
      data += colSizes[col];
    }

    colSizes[col] = htonl(colSizes[col]);
  }

  *actualLen = *dataLen;
  *groupId = pBlock->info.groupId;
}

const char* blockCompressDecode(SSDataBlock* pBlock, int32_t numOfCols, int32_t numOfRows, const char* pData) {
  blockDataEnsureCapacity(pBlock, numOfRows);
1793

1794 1795 1796 1797 1798 1799 1800 1801
  const char* pStart = pData;

  int32_t dataLen = *(int32_t*)pStart;
  pStart += sizeof(int32_t);

  pBlock->info.groupId = *(uint64_t*)pStart;
  pStart += sizeof(uint64_t);

L
Liu Jicong 已提交
1802 1803 1804 1805 1806
  if (pBlock->pDataBlock == NULL) {
    pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
    taosArraySetSize(pBlock->pDataBlock, numOfCols);
  }

1807 1808
  pBlock->info.numOfCols = taosArrayGetSize(pBlock->pDataBlock);
  ASSERT(pBlock->pDataBlock->size >= numOfCols);
L
Liu Jicong 已提交
1809 1810

  for (int32_t i = 0; i < numOfCols; ++i) {
1811 1812 1813 1814 1815 1816
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
    pColInfoData->info.type = *(int16_t*)pStart;
    pStart += sizeof(int16_t);

    pColInfoData->info.bytes = *(int32_t*)pStart;
    pStart += sizeof(int32_t);
1817 1818 1819 1820

    if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
      pBlock->info.hasVarCol = true;
    }
1821 1822 1823 1824
  }

  blockDataEnsureCapacity(pBlock, numOfRows);

1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840
  int32_t* colLen = (int32_t*)pStart;
  pStart += sizeof(int32_t) * numOfCols;

  for (int32_t i = 0; i < numOfCols; ++i) {
    colLen[i] = htonl(colLen[i]);
    ASSERT(colLen[i] >= 0);

    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
    if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
      pColInfoData->varmeta.length = colLen[i];
      pColInfoData->varmeta.allocLen = colLen[i];

      memcpy(pColInfoData->varmeta.offset, pStart, sizeof(int32_t) * numOfRows);
      pStart += sizeof(int32_t) * numOfRows;

      if (colLen[i] > 0) {
H
Haojun Liao 已提交
1841
        taosMemoryFreeClear(pColInfoData->pData);
1842 1843 1844
        pColInfoData->pData = taosMemoryMalloc(colLen[i]);
      }
    } else {
L
Liu Jicong 已提交
1845 1846 1847
      if (pColInfoData->nullbitmap == NULL) {
        pColInfoData->nullbitmap = taosMemoryCalloc(1, BitmapLen(numOfRows));
      }
1848 1849 1850 1851 1852
      memcpy(pColInfoData->nullbitmap, pStart, BitmapLen(numOfRows));
      pStart += BitmapLen(numOfRows);
    }

    if (colLen[i] > 0) {
L
Liu Jicong 已提交
1853 1854 1855
      if (pColInfoData->pData == NULL) {
        pColInfoData->pData = taosMemoryCalloc(1, colLen[i]);
      }
1856 1857 1858 1859 1860 1861 1862 1863 1864 1865
      memcpy(pColInfoData->pData, pStart, colLen[i]);
    }

    // TODO
    // setting this flag to true temporarily so aggregate function on stable will
    // examine NULL value for non-primary key column
    pColInfoData->hasNull = true;
    pStart += colLen[i];
  }

1866
  pBlock->info.rows = numOfRows;
1867 1868
  ASSERT(pStart - pData == dataLen);
  return pStart;
L
Liu Jicong 已提交
1869
}