tfill.c 21.9 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * 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 "os.h"
#include "taosdef.h"
H
Hongze Cheng 已提交
18
#include "tmsg.h"
19 20
#include "ttypes.h"

S
common  
Shengliang Guan 已提交
21
#include "tcommon.h"
22
#include "thash.h"
23 24
#include "ttime.h"

25
#include "executorInt.h"
X
Xiaoyu Wang 已提交
26
#include "function.h"
27
#include "querynodes.h"
X
Xiaoyu Wang 已提交
28
#include "tdatablock.h"
29 30
#include "tfill.h"

31
#define FILL_IS_ASC_FILL(_f) ((_f)->order == TSDB_ORDER_ASC)
X
Xiaoyu Wang 已提交
32 33
#define DO_INTERPOLATION(_v1, _v2, _k1, _k2, _k) \
  ((_v1) + ((_v2) - (_v1)) * (((double)(_k)) - ((double)(_k1))) / (((double)(_k2)) - ((double)(_k1))))
34 35

static void setTagsValue(SFillInfo* pFillInfo, void** data, int32_t genRows) {
X
Xiaoyu Wang 已提交
36
  for (int32_t j = 0; j < pFillInfo->numOfCols; ++j) {
37 38 39 40 41
    SFillColInfo* pCol = &pFillInfo->pFillCol[j];
    if (TSDB_COL_IS_NORMAL_COL(pCol->flag) || TSDB_COL_IS_UD_COL(pCol->flag)) {
      continue;
    }

42
    SResSchema* pSchema = &pCol->pExpr->base.resSchema;
X
Xiaoyu Wang 已提交
43
    char*       val1 = elePtrAt(data[j], pSchema->bytes, genRows);
44 45 46

    assert(pCol->tagIndex >= 0 && pCol->tagIndex < pFillInfo->numOfTags);
    SFillTagColInfo* pTag = &pFillInfo->pTags[pCol->tagIndex];
47
    assignVal(val1, pTag->tagVal, pSchema->bytes, pSchema->type);
48 49 50
  }
}

51
static void setNullRow(SSDataBlock* pBlock, int32_t numOfCol, int32_t rowIndex) {
52
  // the first are always the timestamp column, so start from the second column.
53 54 55
  for (int32_t i = 1; i < pBlock->info.numOfCols; ++i) {
    SColumnInfoData* p = taosArrayGet(pBlock->pDataBlock, i);
    colDataAppendNULL(p, rowIndex);
56 57 58
  }
}

X
Xiaoyu Wang 已提交
59
#define GET_DEST_SLOT_ID(_p) ((_p)->pExpr->base.resSchema.slotId)
60 61 62
#define GET_SRC_SLOT_ID(_p)  ((_p)->pExpr->base.pParam[0].pCol->slotId)

static void doSetVal(SColumnInfoData* pDstColInfoData, int32_t rowIndex, const SGroupKeys* pKey);
63

X
Xiaoyu Wang 已提交
64 65 66
static void doFillOneRowResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, SSDataBlock* pSrcBlock, int64_t ts,
                               bool outOfBound) {
  SPoint  point1, point2, point;
67 68 69
  int32_t step = GET_FORWARD_DIRECTION_FACTOR(pFillInfo->order);

  // set the primary timestamp column value
X
Xiaoyu Wang 已提交
70 71 72 73 74
  int32_t          index = pFillInfo->numOfCurrent;
  SColumnInfoData* pCol0 = taosArrayGet(pBlock->pDataBlock, 0);
  char*            val = colDataGetData(pCol0, index);

  *(TSKEY*)val = pFillInfo->currentKey;
75 76 77

  // set the other values
  if (pFillInfo->type == TSDB_FILL_PREV) {
78
    SArray* p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->prev : pFillInfo->next;
79

80 81 82 83
    for (int32_t i = 1; i < pFillInfo->numOfCols; ++i) {
      SFillColInfo* pCol = &pFillInfo->pFillCol[i];
      if (TSDB_COL_IS_TAG(pCol->flag)) {
        continue;
84
      }
85

X
Xiaoyu Wang 已提交
86
      SGroupKeys*      pKey = taosArrayGet(p, i);
87 88
      SColumnInfoData* pDstColInfoData = taosArrayGet(pBlock->pDataBlock, GET_DEST_SLOT_ID(pCol));
      doSetVal(pDstColInfoData, index, pKey);
89 90
    }
  } else if (pFillInfo->type == TSDB_FILL_NEXT) {
91
    SArray* p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->next : pFillInfo->prev;
92

93 94 95 96
    for (int32_t i = 1; i < pFillInfo->numOfCols; ++i) {
      SFillColInfo* pCol = &pFillInfo->pFillCol[i];
      if (TSDB_COL_IS_TAG(pCol->flag)) {
        continue;
97
      }
98

X
Xiaoyu Wang 已提交
99
      SGroupKeys*      pKey = taosArrayGet(p, i);
100 101
      SColumnInfoData* pDstColInfoData = taosArrayGet(pBlock->pDataBlock, GET_DEST_SLOT_ID(pCol));
      doSetVal(pDstColInfoData, index, pKey);
102 103 104
    }
  } else if (pFillInfo->type == TSDB_FILL_LINEAR) {
    // TODO : linear interpolation supports NULL value
105 106 107
    if (outOfBound) {
      setNullRow(pBlock, pFillInfo->numOfCols, index);
    } else {
108 109 110 111 112 113
      for (int32_t i = 1; i < pFillInfo->numOfCols; ++i) {
        SFillColInfo* pCol = &pFillInfo->pFillCol[i];
        if (TSDB_COL_IS_TAG(pCol->flag)) {
          continue;
        }

114 115
        int32_t srcSlotId = GET_SRC_SLOT_ID(pCol);

X
Xiaoyu Wang 已提交
116
        int32_t          dstSlotId = GET_DEST_SLOT_ID(pCol);
117
        SColumnInfoData* pDstCol = taosArrayGet(pBlock->pDataBlock, dstSlotId);
118

X
Xiaoyu Wang 已提交
119
        int16_t     type = pCol->pExpr->base.resSchema.type;
120 121 122
        SGroupKeys* pKey = taosArrayGet(pFillInfo->prev, i);
        if (IS_VAR_DATA_TYPE(type) || type == TSDB_DATA_TYPE_BOOL || pKey->isNull) {
          colDataAppendNULL(pDstCol, index);
123 124 125
          continue;
        }

126
        SGroupKeys* pKey1 = taosArrayGet(pFillInfo->prev, 0);
X
Xiaoyu Wang 已提交
127
        int64_t     prevTs = *(int64_t*)pKey1->pData;
128 129

        SColumnInfoData* pSrcCol = taosArrayGet(pSrcBlock->pDataBlock, srcSlotId);
X
Xiaoyu Wang 已提交
130
        char*            data = colDataGetData(pSrcCol, pFillInfo->index);
131 132 133 134 135 136

        point1 = (SPoint){.key = prevTs, .val = pKey->pData};
        point2 = (SPoint){.key = ts, .val = data};

        int64_t out = 0;
        point = (SPoint){.key = pFillInfo->currentKey, .val = &out};
137
        taosGetLinearInterpolationVal(&point, type, &point1, &point2, type);
138 139

        colDataAppend(pDstCol, index, (const char*)&out, false);
140 141
      }
    }
142 143
  } else if (pFillInfo->type == TSDB_FILL_NULL) {  // fill with NULL
    setNullRow(pBlock, pFillInfo->numOfCols, index);
X
Xiaoyu Wang 已提交
144
  } else {  // fill with user specified value for each column
145 146
    for (int32_t i = 1; i < pFillInfo->numOfCols; ++i) {
      SFillColInfo* pCol = &pFillInfo->pFillCol[i];
X
Xiaoyu Wang 已提交
147
      if (TSDB_COL_IS_TAG(pCol->flag) /* || IS_VAR_DATA_TYPE(pCol->schema.type)*/) {
148 149 150
        continue;
      }

151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166
      SVariant* pVar = &pFillInfo->pFillCol[i].fillVal;

      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, i);
      if (pDst->info.type == TSDB_DATA_TYPE_FLOAT) {
        float v = 0;
        GET_TYPED_DATA(v, float, pVar->nType, &pVar->i);
        colDataAppend(pDst, index, (char*)&v, false);
      } else if (pDst->info.type == TSDB_DATA_TYPE_DOUBLE) {
        double v = 0;
        GET_TYPED_DATA(v, double, pVar->nType, &pVar->i);
        colDataAppend(pDst, index, (char*)&v, false);
      } else if (IS_SIGNED_NUMERIC_TYPE(pDst->info.type)) {
        int64_t v = 0;
        GET_TYPED_DATA(v, int64_t, pVar->nType, &pVar->i);
        colDataAppend(pDst, index, (char*)&v, false);
      }
167 168 169
    }
  }

X
Xiaoyu Wang 已提交
170
  //  setTagsValue(pFillInfo, data, index);
171
  SInterval* pInterval = &pFillInfo->interval;
X
Xiaoyu Wang 已提交
172 173
  pFillInfo->currentKey =
      taosTimeAdd(pFillInfo->currentKey, pInterval->sliding * step, pInterval->slidingUnit, pInterval->precision);
174 175 176
  pFillInfo->numOfCurrent++;
}

177 178 179 180 181 182 183 184 185 186
void doSetVal(SColumnInfoData* pDstCol, int32_t rowIndex, const SGroupKeys* pKey) {
  if (pKey->isNull) {
    colDataAppendNULL(pDstCol, rowIndex);
  } else {
    colDataAppend(pDstCol, rowIndex, pKey->pData, false);
  }
}

static void initBeforeAfterDataBuf(SFillInfo* pFillInfo) {
  if (taosArrayGetSize(pFillInfo->next) > 0) {
187 188 189
    return;
  }

190
  for (int i = 0; i < pFillInfo->numOfCols; i++) {
191
    SFillColInfo* pCol = &pFillInfo->pFillCol[i];
192

X
Xiaoyu Wang 已提交
193
    SGroupKeys  key = {0};
194
    SResSchema* pSchema = &pCol->pExpr->base.resSchema;
X
Xiaoyu Wang 已提交
195
    key.pData = taosMemoryMalloc(pSchema->bytes);
196 197
    key.isNull = true;
    key.bytes = pSchema->bytes;
X
Xiaoyu Wang 已提交
198
    key.type = pSchema->type;
199 200 201 202 203

    taosArrayPush(pFillInfo->next, &key);

    key.pData = taosMemoryMalloc(pSchema->bytes);
    taosArrayPush(pFillInfo->prev, &key);
204 205 206
  }
}

207 208 209
static void saveColData(SArray* rowBuf, int32_t columnIndex, const char* src, bool isNull);

static void copyCurrentRowIntoBuf(SFillInfo* pFillInfo, int32_t rowIndex, SArray* pRow) {
210
  for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
211 212 213 214
    int32_t srcSlotId = GET_SRC_SLOT_ID(&pFillInfo->pFillCol[i]);

    SColumnInfoData* pSrcCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, srcSlotId);

X
Xiaoyu Wang 已提交
215
    bool  isNull = colDataIsNull_s(pSrcCol, rowIndex);
216 217
    char* p = colDataGetData(pSrcCol, rowIndex);
    saveColData(pRow, i, p, isNull);
218 219 220
  }
}

221
static int32_t fillResultImpl(SFillInfo* pFillInfo, SSDataBlock* pBlock, int32_t outputRows) {
222 223
  pFillInfo->numOfCurrent = 0;

224 225
  // todo make sure the first column is always the primary timestamp column?
  SColumnInfoData* pTsCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, 0);
226 227

  int32_t step = GET_FORWARD_DIRECTION_FACTOR(pFillInfo->order);
X
Xiaoyu Wang 已提交
228
  bool    ascFill = FILL_IS_ASC_FILL(pFillInfo);
229

230 231 232
#if 0
  ASSERT(ascFill && (pFillInfo->currentKey >= pFillInfo->start) || (!ascFill && (pFillInfo->currentKey <= pFillInfo->start)));
#endif
233 234

  while (pFillInfo->numOfCurrent < outputRows) {
235
    int64_t ts = ((int64_t*)pTsCol->pData)[pFillInfo->index];
236 237

    // set the next value for interpolation
238 239
    if ((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) {
      copyCurrentRowIntoBuf(pFillInfo, pFillInfo->index, pFillInfo->next);
240 241
    }

X
Xiaoyu Wang 已提交
242 243
    if (((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) &&
        pFillInfo->numOfCurrent < outputRows) {
244
      // fill the gap between two input rows
X
Xiaoyu Wang 已提交
245 246
      while (((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) &&
             pFillInfo->numOfCurrent < outputRows) {
247
        doFillOneRowResult(pFillInfo, pBlock, pFillInfo->pSrcBlock, ts, false);
248 249 250 251 252 253 254 255 256
      }

      // output buffer is full, abort
      if (pFillInfo->numOfCurrent == outputRows) {
        pFillInfo->numOfTotal += pFillInfo->numOfCurrent;
        return outputRows;
      }
    } else {
      assert(pFillInfo->currentKey == ts);
257

258 259
      if (pFillInfo->type == TSDB_FILL_NEXT && (pFillInfo->index + 1) < pFillInfo->numOfRows) {
        ++pFillInfo->index;
260
        copyCurrentRowIntoBuf(pFillInfo, pFillInfo->index, pFillInfo->next);
261 262 263 264 265 266
        --pFillInfo->index;
      }

      // assign rows to dst buffer
      for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
        SFillColInfo* pCol = &pFillInfo->pFillCol[i];
X
Xiaoyu Wang 已提交
267
        if (TSDB_COL_IS_TAG(pCol->flag) /* || IS_VAR_DATA_TYPE(pCol->schema.type)*/) {
268 269 270
          continue;
        }

271 272
        int32_t srcSlotId = GET_SRC_SLOT_ID(pCol);
        int32_t dstSlotId = GET_DEST_SLOT_ID(pCol);
273

274 275 276 277 278 279 280 281 282
        SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, dstSlotId);
        SColumnInfoData* pSrc = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, srcSlotId);

        char* src = colDataGetData(pSrc, pFillInfo->index);
        if (i == 0 || (/*pCol->functionId != FUNCTION_COUNT &&*/ !colDataIsNull_s(pSrc, pFillInfo->index)) /*||
            (pCol->functionId == FUNCTION_COUNT && GET_INT64_VAL(src) != 0)*/) {
          bool isNull = colDataIsNull_s(pSrc, pFillInfo->index);
          colDataAppend(pDst, pFillInfo->numOfCurrent, src, isNull);
          saveColData(pFillInfo->prev, i, src, isNull);
283 284
        } else {  // i > 0 and data is null , do interpolation
          if (pFillInfo->type == TSDB_FILL_PREV) {
X
Xiaoyu Wang 已提交
285
            SGroupKeys* pKey = taosArrayGet(pFillInfo->prev, i);
286
            doSetVal(pDst, pFillInfo->numOfCurrent, pKey);
287
          } else if (pFillInfo->type == TSDB_FILL_LINEAR) {
288 289 290 291 292
            bool isNull = colDataIsNull_s(pSrc, pFillInfo->index);
            colDataAppend(pDst, pFillInfo->numOfCurrent, src, isNull);
            saveColData(pFillInfo->prev, i, src, isNull);
          } else if (pFillInfo->type == TSDB_FILL_NULL) {
            colDataAppendNULL(pDst, pFillInfo->numOfCurrent);
293
          } else if (pFillInfo->type == TSDB_FILL_NEXT) {
X
Xiaoyu Wang 已提交
294
            SGroupKeys* pKey = taosArrayGet(pFillInfo->next, i);
295
            doSetVal(pDst, pFillInfo->numOfCurrent, pKey);
296
          } else {
297 298
            SVariant* pVar = &pFillInfo->pFillCol[i].fillVal;
            colDataAppend(pDst, pFillInfo->numOfCurrent, (char*)&pVar->i, false);
299 300 301 302 303
          }
        }
      }

      // set the tag value for final result
X
Xiaoyu Wang 已提交
304 305 306 307
      //      setTagsValue(pFillInfo, data, pFillInfo->numOfCurrent);
      SInterval* pInterval = &pFillInfo->interval;
      pFillInfo->currentKey =
          taosTimeAdd(pFillInfo->currentKey, pInterval->sliding * step, pInterval->slidingUnit, pInterval->precision);
308 309 310 311 312 313 314

      pFillInfo->index += 1;
      pFillInfo->numOfCurrent += 1;
    }

    if (pFillInfo->index >= pFillInfo->numOfRows || pFillInfo->numOfCurrent >= outputRows) {
      /* the raw data block is exhausted, next value does not exists */
X
Xiaoyu Wang 已提交
315 316 317
      //      if (pFillInfo->index >= pFillInfo->numOfRows) {
      //        taosMemoryFreeClear(*next);
      //      }
318 319 320 321 322 323 324 325
      pFillInfo->numOfTotal += pFillInfo->numOfCurrent;
      return pFillInfo->numOfCurrent;
    }
  }

  return pFillInfo->numOfCurrent;
}

326
static void saveColData(SArray* rowBuf, int32_t columnIndex, const char* src, bool isNull) {
X
Xiaoyu Wang 已提交
327
  SGroupKeys* pKey = taosArrayGet(rowBuf, columnIndex);
328 329 330 331 332 333 334 335 336
  if (isNull) {
    pKey->isNull = true;
  } else {
    memcpy(pKey->pData, src, pKey->bytes);
    pKey->isNull = false;
  }
}

static int64_t appendFilledResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, int64_t resultCapacity) {
337 338 339 340 341 342
  /*
   * These data are generated according to fill strategy, since the current timestamp is out of the time window of
   * real result set. Note that we need to keep the direct previous result rows, to generated the filled data.
   */
  pFillInfo->numOfCurrent = 0;
  while (pFillInfo->numOfCurrent < resultCapacity) {
343
    doFillOneRowResult(pFillInfo, pBlock, pFillInfo->pSrcBlock, pFillInfo->start, true);
344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359
  }

  pFillInfo->numOfTotal += pFillInfo->numOfCurrent;

  assert(pFillInfo->numOfCurrent == resultCapacity);
  return resultCapacity;
}

// there are no duplicated tags in the SFillTagColInfo list
static int32_t setTagColumnInfo(SFillInfo* pFillInfo, int32_t numOfCols, int32_t capacity) {
  int32_t rowsize = 0;
  int32_t numOfTags = 0;

  int32_t k = 0;
  for (int32_t i = 0; i < numOfCols; ++i) {
    SFillColInfo* pColInfo = &pFillInfo->pFillCol[i];
X
Xiaoyu Wang 已提交
360
    SResSchema*   pSchema = &pColInfo->pExpr->base.resSchema;
361

362
    if (TSDB_COL_IS_TAG(pColInfo->flag) || pSchema->type == TSDB_DATA_TYPE_BINARY) {
363 364
      numOfTags += 1;

X
Xiaoyu Wang 已提交
365
      bool    exists = false;
366 367
      int32_t index = -1;
      for (int32_t j = 0; j < k; ++j) {
368
        if (pFillInfo->pTags[j].col.colId == pSchema->slotId) {
369 370 371 372 373 374 375
          exists = true;
          index = j;
          break;
        }
      }

      if (!exists) {
376 377
        SSchema* pSchema1 = &pFillInfo->pTags[k].col;
        pSchema1->colId = pSchema->slotId;
X
Xiaoyu Wang 已提交
378
        pSchema1->type = pSchema->type;
379
        pSchema1->bytes = pSchema->bytes;
380

381
        pFillInfo->pTags[k].tagVal = taosMemoryCalloc(1, pSchema->bytes);
382 383 384 385 386 387 388 389
        pColInfo->tagIndex = k;

        k += 1;
      } else {
        pColInfo->tagIndex = index;
      }
    }

390
    rowsize += pSchema->bytes;
391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406
  }

  pFillInfo->numOfTags = numOfTags;

  assert(k <= pFillInfo->numOfTags);
  return rowsize;
}

static int32_t taosNumOfRemainRows(SFillInfo* pFillInfo) {
  if (pFillInfo->numOfRows == 0 || (pFillInfo->numOfRows > 0 && pFillInfo->index >= pFillInfo->numOfRows)) {
    return 0;
  }

  return pFillInfo->numOfRows - pFillInfo->index;
}

407
struct SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
X
Xiaoyu Wang 已提交
408 409
                                     SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol,
                                     const char* id) {
410 411 412 413
  if (fillType == TSDB_FILL_NONE) {
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
414
  SFillInfo* pFillInfo = taosMemoryCalloc(1, sizeof(SFillInfo));
415 416 417 418 419
  if (pFillInfo == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

420
  taosResetFillInfo(pFillInfo, skey);
421 422
  pFillInfo->order = order;

X
Xiaoyu Wang 已提交
423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441
  switch (fillType) {
    case FILL_MODE_NONE:
      pFillInfo->type = TSDB_FILL_NONE;
      break;
    case FILL_MODE_PREV:
      pFillInfo->type = TSDB_FILL_PREV;
      break;
    case FILL_MODE_NULL:
      pFillInfo->type = TSDB_FILL_NULL;
      break;
    case FILL_MODE_LINEAR:
      pFillInfo->type = TSDB_FILL_LINEAR;
      break;
    case FILL_MODE_NEXT:
      pFillInfo->type = TSDB_FILL_NEXT;
      break;
    case FILL_MODE_VALUE:
      pFillInfo->type = TSDB_FILL_SET_VALUE;
      break;
442 443 444 445 446
    default:
      terrno = TSDB_CODE_INVALID_PARA;
      return NULL;
  }

X
Xiaoyu Wang 已提交
447 448
  pFillInfo->type = fillType;
  pFillInfo->pFillCol = pCol;
449 450
  pFillInfo->numOfTags = numOfTags;
  pFillInfo->numOfCols = numOfCols;
X
Xiaoyu Wang 已提交
451 452 453 454 455 456 457 458 459 460
  pFillInfo->alloc = capacity;
  pFillInfo->id = id;
  pFillInfo->interval = *pInterval;

  //  if (numOfTags > 0) {
  pFillInfo->pTags = taosMemoryCalloc(numOfCols, sizeof(SFillTagColInfo));
  for (int32_t i = 0; i < numOfCols; ++i) {
    pFillInfo->pTags[i].col.colId = -2;  // TODO
  }
  //  }
461

462 463 464 465 466
  pFillInfo->next = taosArrayInit(numOfCols, sizeof(SGroupKeys));
  pFillInfo->prev = taosArrayInit(numOfCols, sizeof(SGroupKeys));

  initBeforeAfterDataBuf(pFillInfo);

467 468 469 470 471 472
  pFillInfo->rowSize = setTagColumnInfo(pFillInfo, pFillInfo->numOfCols, pFillInfo->alloc);
  assert(pFillInfo->rowSize > 0);
  return pFillInfo;
}

void taosResetFillInfo(SFillInfo* pFillInfo, TSKEY startTimestamp) {
X
Xiaoyu Wang 已提交
473 474 475 476 477
  pFillInfo->start = startTimestamp;
  pFillInfo->currentKey = startTimestamp;
  pFillInfo->end = startTimestamp;
  pFillInfo->index = -1;
  pFillInfo->numOfRows = 0;
478
  pFillInfo->numOfCurrent = 0;
X
Xiaoyu Wang 已提交
479
  pFillInfo->numOfTotal = 0;
480 481 482 483 484 485 486
}

void* taosDestroyFillInfo(SFillInfo* pFillInfo) {
  if (pFillInfo == NULL) {
    return NULL;
  }

487 488
  taosArrayDestroy(pFillInfo->prev);
  taosArrayDestroy(pFillInfo->next);
489

X
Xiaoyu Wang 已提交
490
  for (int32_t i = 0; i < pFillInfo->numOfTags; ++i) {
wafwerar's avatar
wafwerar 已提交
491
    taosMemoryFreeClear(pFillInfo->pTags[i].tagVal);
492 493
  }

wafwerar's avatar
wafwerar 已提交
494 495 496
  taosMemoryFreeClear(pFillInfo->pTags);
  taosMemoryFreeClear(pFillInfo->pFillCol);
  taosMemoryFreeClear(pFillInfo);
497 498 499 500 501 502 503 504 505 506
  return NULL;
}

void taosFillSetStartInfo(SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey) {
  if (pFillInfo->type == TSDB_FILL_NONE) {
    return;
  }

  pFillInfo->end = endKey;
  if (!FILL_IS_ASC_FILL(pFillInfo)) {
507
    pFillInfo->end = taosTimeTruncate(endKey, &pFillInfo->interval, pFillInfo->interval.precision);
508 509
  }

X
Xiaoyu Wang 已提交
510
  pFillInfo->index = 0;
511 512 513 514
  pFillInfo->numOfRows = numOfRows;
}

void taosFillSetInputDataBlock(SFillInfo* pFillInfo, const SSDataBlock* pInput) {
X
Xiaoyu Wang 已提交
515
  pFillInfo->pSrcBlock = (SSDataBlock*)pInput;
516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532
}

bool taosFillHasMoreResults(SFillInfo* pFillInfo) {
  int32_t remain = taosNumOfRemainRows(pFillInfo);
  if (remain > 0) {
    return true;
  }

  if (pFillInfo->numOfTotal > 0 && (((pFillInfo->end > pFillInfo->start) && FILL_IS_ASC_FILL(pFillInfo)) ||
                                    (pFillInfo->end < pFillInfo->start && !FILL_IS_ASC_FILL(pFillInfo)))) {
    return getNumOfResultsAfterFillGap(pFillInfo, pFillInfo->end, 4096) > 0;
  }

  return false;
}

int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, TSKEY ekey, int32_t maxNumOfRows) {
533
  SColumnInfoData* pCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, 0);
534

X
Xiaoyu Wang 已提交
535 536
  int64_t* tsList = (int64_t*)pCol->pData;
  int32_t  numOfRows = taosNumOfRemainRows(pFillInfo);
537 538 539

  TSKEY ekey1 = ekey;
  if (!FILL_IS_ASC_FILL(pFillInfo)) {
540
    pFillInfo->end = taosTimeTruncate(ekey, &pFillInfo->interval, pFillInfo->interval.precision);
541 542 543 544 545
  }

  int64_t numOfRes = -1;
  if (numOfRows > 0) {  // still fill gap within current data block, not generating data after the result set.
    TSKEY lastKey = tsList[pFillInfo->numOfRows - 1];
X
Xiaoyu Wang 已提交
546 547
    numOfRes = taosTimeCountInterval(lastKey, pFillInfo->currentKey, pFillInfo->interval.sliding,
                                     pFillInfo->interval.slidingUnit, pFillInfo->interval.precision);
548 549
    numOfRes += 1;
    assert(numOfRes >= numOfRows);
X
Xiaoyu Wang 已提交
550
  } else {  // reach the end of data
551 552 553 554
    if ((ekey1 < pFillInfo->currentKey && FILL_IS_ASC_FILL(pFillInfo)) ||
        (ekey1 > pFillInfo->currentKey && !FILL_IS_ASC_FILL(pFillInfo))) {
      return 0;
    }
X
Xiaoyu Wang 已提交
555 556
    numOfRes = taosTimeCountInterval(ekey1, pFillInfo->currentKey, pFillInfo->interval.sliding,
                                     pFillInfo->interval.slidingUnit, pFillInfo->interval.precision);
557 558 559 560 561 562
    numOfRes += 1;
  }

  return (numOfRes > maxNumOfRows) ? maxNumOfRows : numOfRes;
}

X
Xiaoyu Wang 已提交
563 564
int32_t taosGetLinearInterpolationVal(SPoint* point, int32_t outputType, SPoint* point1, SPoint* point2,
                                      int32_t inputType) {
565 566 567 568 569 570 571 572 573 574
  double v1 = -1, v2 = -1;
  GET_TYPED_DATA(v1, double, inputType, point1->val);
  GET_TYPED_DATA(v2, double, inputType, point2->val);

  double r = DO_INTERPOLATION(v1, v2, point1->key, point2->key, point->key);
  SET_TYPED_DATA(point->val, outputType, r);

  return TSDB_CODE_SUCCESS;
}

575
int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, SSDataBlock* p, int32_t capacity) {
576 577 578 579 580 581 582
  int32_t remain = taosNumOfRemainRows(pFillInfo);

  int64_t numOfRes = getNumOfResultsAfterFillGap(pFillInfo, pFillInfo->end, capacity);
  assert(numOfRes <= capacity);

  // no data existed for fill operation now, append result according to the fill strategy
  if (remain == 0) {
583
    appendFilledResult(pFillInfo, p, numOfRes);
584
  } else {
X
Xiaoyu Wang 已提交
585
    fillResultImpl(pFillInfo, p, (int32_t)numOfRes);
586 587 588
    assert(numOfRes == pFillInfo->numOfCurrent);
  }

X
Xiaoyu Wang 已提交
589 590 591 592 593
  //  qDebug("fill:%p, generated fill result, src block:%d, index:%d, brange:%"PRId64"-%"PRId64", currentKey:%"PRId64",
  //  current:%d, total:%d, %p",
  //      pFillInfo, pFillInfo->numOfRows, pFillInfo->index, pFillInfo->start, pFillInfo->end, pFillInfo->currentKey,
  //      pFillInfo->numOfCurrent,
  //         pFillInfo->numOfTotal, pFillInfo->handle);
594 595 596

  return numOfRes;
}
597

X
Xiaoyu Wang 已提交
598
int64_t getFillInfoStart(struct SFillInfo* pFillInfo) { return pFillInfo->start; }
599

600 601
SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, const struct SNodeListNode* pValNode) {
  SFillColInfo* pFillCol = taosMemoryCalloc(numOfOutput, sizeof(SFillColInfo));
602 603 604 605
  if (pFillCol == NULL) {
    return NULL;
  }

X
Xiaoyu Wang 已提交
606 607
  size_t len = (pValNode != NULL) ? LIST_LENGTH(pValNode->pNodeList) : 0;
  for (int32_t i = 0; i < numOfOutput; ++i) {
608 609 610
    SExprInfo* pExprInfo = &pExpr[i];
    pFillCol[i].pExpr = pExprInfo;
    pFillCol[i].tagIndex = -2;
611

612 613 614
    // todo refactor
    if (len > 0) {
      // if the user specified value is less than the column, alway use the last one as the fill value
X
Xiaoyu Wang 已提交
615
      int32_t index = (i >= len) ? (len - 1) : i;
616 617

      SValueNode* pv = (SValueNode*)nodesListGetNode(pValNode->pNodeList, index);
X
Xiaoyu Wang 已提交
618
      nodesValueNodeToVariant(pv, &pFillCol[i].fillVal);
619
    }
620 621

    if (pExprInfo->base.numOfParams > 0) {
X
Xiaoyu Wang 已提交
622
      pFillCol[i].flag = pExprInfo->base.pParam[0].pCol->flag;  // always be the normal column for table query
623
    }
624 625 626 627
  }

  return pFillCol;
}