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

#include "os.h"
17
#include "query.h"
18
#include "taosdef.h"
H
Hongze Cheng 已提交
19
#include "tmsg.h"
20 21
#include "ttypes.h"

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

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

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

H
Haojun Liao 已提交
36 37 38 39 40 41 42 43 44 45 46 47 48
#define GET_DEST_SLOT_ID(_p) ((_p)->pExpr->base.resSchema.slotId)

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

static void setNullRow(SSDataBlock* pBlock, SFillInfo* pFillInfo, int32_t rowIndex) {
  for(int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
    SFillColInfo* pCol = &pFillInfo->pFillCol[i];
    int32_t dstSlotId = GET_DEST_SLOT_ID(pCol);
    SColumnInfoData* pDstColInfo = taosArrayGet(pBlock->pDataBlock, dstSlotId);
    if (pCol->notFillCol) {
      if (pDstColInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
        colDataAppend(pDstColInfo, rowIndex, (const char*)&pFillInfo->currentKey, false);
      } else {
H
Haojun Liao 已提交
49
        SArray* p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->prev.pRowVal : pFillInfo->next.pRowVal;
H
Haojun Liao 已提交
50 51 52
        SGroupKeys* pKey = taosArrayGet(p, i);
        doSetVal(pDstColInfo, rowIndex, pKey);
      }
53
    } else {
H
Haojun Liao 已提交
54
      colDataAppendNULL(pDstColInfo, rowIndex);
55
    }
56 57 58
  }
}

59
static void doSetUserSpecifiedValue(SColumnInfoData* pDst, SVariant* pVar, int32_t rowIndex, int64_t currentKey) {
60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78
  if (pDst->info.type == TSDB_DATA_TYPE_FLOAT) {
    float v = 0;
    GET_TYPED_DATA(v, float, pVar->nType, &pVar->i);
    colDataAppend(pDst, rowIndex, (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, rowIndex, (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, rowIndex, (char*)&v, false);
  } else if (pDst->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
    colDataAppend(pDst, rowIndex, (const char*)&currentKey, false);
  } else {  // varchar/nchar data
    colDataAppendNULL(pDst, rowIndex);
  }
}

79 80
static void doFillOneRow(SFillInfo* pFillInfo, SSDataBlock* pBlock, SSDataBlock* pSrcBlock, int64_t ts,
                         bool outOfBound) {
X
Xiaoyu Wang 已提交
81
  SPoint  point1, point2, point;
82 83
  int32_t step = GET_FORWARD_DIRECTION_FACTOR(pFillInfo->order);

84
  // set the primary timestamp column value
85
  int32_t index = pBlock->info.rows;
86 87 88

  // set the other values
  if (pFillInfo->type == TSDB_FILL_PREV) {
H
Haojun Liao 已提交
89
    SArray* p = FILL_IS_ASC_FILL(pFillInfo)? pFillInfo->prev.pRowVal : pFillInfo->next.pRowVal;
90

91
    for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
92 93 94
      SFillColInfo* pCol = &pFillInfo->pFillCol[i];

      SColumnInfoData* pDstColInfoData = taosArrayGet(pBlock->pDataBlock, GET_DEST_SLOT_ID(pCol));
95 96

      if (pDstColInfoData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
97
        colDataAppend(pDstColInfoData, index, (const char*)&pFillInfo->currentKey, false);
98 99 100 101
      } else {
        SGroupKeys* pKey = taosArrayGet(p, i);
        doSetVal(pDstColInfoData, index, pKey);
      }
102 103
    }
  } else if (pFillInfo->type == TSDB_FILL_NEXT) {
H
Haojun Liao 已提交
104
    SArray* p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->next.pRowVal : pFillInfo->prev.pRowVal;
105
    // todo  refactor: start from 0 not 1
106
    for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
107 108
      SFillColInfo* pCol = &pFillInfo->pFillCol[i];
      SColumnInfoData* pDstColInfoData = taosArrayGet(pBlock->pDataBlock, GET_DEST_SLOT_ID(pCol));
109 110 111 112 113 114 115

      if (pDstColInfoData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
        colDataAppend(pDstColInfoData, index, (const char*)&pFillInfo->currentKey, false);
      } else {
        SGroupKeys* pKey = taosArrayGet(p, i);
        doSetVal(pDstColInfoData, index, pKey);
      }
116 117 118
    }
  } else if (pFillInfo->type == TSDB_FILL_LINEAR) {
    // TODO : linear interpolation supports NULL value
119
    if (outOfBound) {
H
Haojun Liao 已提交
120
      setNullRow(pBlock, pFillInfo, index);
121
    } else {
122
      for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
123 124
        SFillColInfo* pCol = &pFillInfo->pFillCol[i];

X
Xiaoyu Wang 已提交
125
        int32_t          dstSlotId = GET_DEST_SLOT_ID(pCol);
126
        SColumnInfoData* pDstCol = taosArrayGet(pBlock->pDataBlock, dstSlotId);
H
Haojun Liao 已提交
127 128 129 130 131 132
        int16_t          type = pDstCol->info.type;

        if (pCol->notFillCol) {
          if (type == TSDB_DATA_TYPE_TIMESTAMP) {
            colDataAppend(pDstCol, index, (const char*)&pFillInfo->currentKey, false);
          } else {
H
Haojun Liao 已提交
133
            SArray* p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->prev.pRowVal : pFillInfo->next.pRowVal;
H
Haojun Liao 已提交
134 135 136 137
            SGroupKeys* pKey = taosArrayGet(p, i);
            doSetVal(pDstCol, index, pKey);
          }
        } else {
H
Haojun Liao 已提交
138
          SGroupKeys* pKey = taosArrayGet(pFillInfo->prev.pRowVal, i);
H
Haojun Liao 已提交
139 140 141 142
          if (IS_VAR_DATA_TYPE(type) || type == TSDB_DATA_TYPE_BOOL || pKey->isNull) {
            colDataAppendNULL(pDstCol, index);
            continue;
          }
143

H
Haojun Liao 已提交
144
          SGroupKeys* pKey1 = taosArrayGet(pFillInfo->prev.pRowVal, pFillInfo->tsSlotId);
145

H
Haojun Liao 已提交
146
          int64_t prevTs = *(int64_t*)pKey1->pData;
H
Haojun Liao 已提交
147
          int32_t srcSlotId = GET_DEST_SLOT_ID(pCol);
148

H
Haojun Liao 已提交
149 150
          SColumnInfoData* pSrcCol = taosArrayGet(pSrcBlock->pDataBlock, srcSlotId);
          char*            data = colDataGetData(pSrcCol, pFillInfo->index);
151

H
Haojun Liao 已提交
152 153
          point1 = (SPoint){.key = prevTs, .val = pKey->pData};
          point2 = (SPoint){.key = ts, .val = data};
154

H
Haojun Liao 已提交
155 156 157
          int64_t out = 0;
          point = (SPoint){.key = pFillInfo->currentKey, .val = &out};
          taosGetLinearInterpolationVal(&point, type, &point1, &point2, type);
158

H
Haojun Liao 已提交
159 160
          colDataAppend(pDstCol, index, (const char*)&out, false);
        }
161 162
      }
    }
163
  } else if (pFillInfo->type == TSDB_FILL_NULL) {  // fill with NULL
H
Haojun Liao 已提交
164
    setNullRow(pBlock, pFillInfo, index);
X
Xiaoyu Wang 已提交
165
  } else {  // fill with user specified value for each column
166
    for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
167 168
      SFillColInfo* pCol = &pFillInfo->pFillCol[i];

H
Haojun Liao 已提交
169 170 171 172 173 174 175
      int32_t slotId = GET_DEST_SLOT_ID(pCol);
      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, slotId);

      if (pCol->notFillCol) {
        if (pDst->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
          colDataAppend(pDst, index, (const char*)&pFillInfo->currentKey, false);
        } else {
H
Haojun Liao 已提交
176
          SArray* p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->prev.pRowVal : pFillInfo->next.pRowVal;
H
Haojun Liao 已提交
177 178 179 180 181 182 183
          SGroupKeys* pKey = taosArrayGet(p, i);
          doSetVal(pDst, index, pKey);
        }
      } else {
        SVariant* pVar = &pFillInfo->pFillCol[i].fillVal;
        doSetUserSpecifiedValue(pDst, pVar, index, pFillInfo->currentKey);
      }
184 185 186
    }
  }

X
Xiaoyu Wang 已提交
187
  //  setTagsValue(pFillInfo, data, index);
188
  SInterval* pInterval = &pFillInfo->interval;
X
Xiaoyu Wang 已提交
189 190
  pFillInfo->currentKey =
      taosTimeAdd(pFillInfo->currentKey, pInterval->sliding * step, pInterval->slidingUnit, pInterval->precision);
191
  pBlock->info.rows += 1;
192 193 194
  pFillInfo->numOfCurrent++;
}

195 196 197 198 199 200 201 202 203
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) {
H
Haojun Liao 已提交
204
  if (taosArrayGetSize(pFillInfo->next.pRowVal) > 0) {
205 206 207
    return;
  }

208
  for (int i = 0; i < pFillInfo->numOfCols; i++) {
209
    SFillColInfo* pCol = &pFillInfo->pFillCol[i];
210

X
Xiaoyu Wang 已提交
211
    SGroupKeys  key = {0};
212
    SResSchema* pSchema = &pCol->pExpr->base.resSchema;
X
Xiaoyu Wang 已提交
213
    key.pData = taosMemoryMalloc(pSchema->bytes);
214 215
    key.isNull = true;
    key.bytes = pSchema->bytes;
X
Xiaoyu Wang 已提交
216
    key.type = pSchema->type;
217

H
Haojun Liao 已提交
218
    taosArrayPush(pFillInfo->next.pRowVal, &key);
219 220

    key.pData = taosMemoryMalloc(pSchema->bytes);
H
Haojun Liao 已提交
221
    taosArrayPush(pFillInfo->prev.pRowVal, &key);
222 223 224
  }
}

225 226 227
static void saveColData(SArray* rowBuf, int32_t columnIndex, const char* src, bool isNull);

static void copyCurrentRowIntoBuf(SFillInfo* pFillInfo, int32_t rowIndex, SArray* pRow) {
228
  for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
H
Haojun Liao 已提交
229 230 231
    int32_t type = pFillInfo->pFillCol[i].pExpr->pExpr->nodeType;
    if (type == QUERY_NODE_COLUMN) {
      int32_t srcSlotId = GET_DEST_SLOT_ID(&pFillInfo->pFillCol[i]);
232

H
Haojun Liao 已提交
233
      SColumnInfoData* pSrcCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, srcSlotId);
234

H
Haojun Liao 已提交
235 236 237 238 239 240 241 242 243 244 245 246
      bool  isNull = colDataIsNull_s(pSrcCol, rowIndex);
      char* p = colDataGetData(pSrcCol, rowIndex);
      saveColData(pRow, i, p, isNull);
    } else if (type == QUERY_NODE_OPERATOR) {
      SColumnInfoData* pSrcCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, i);

      bool             isNull = colDataIsNull_s(pSrcCol, rowIndex);
      char*            p = colDataGetData(pSrcCol, rowIndex);
      saveColData(pRow, i, p, isNull);
    } else {
      ASSERT(0);
    }
247 248 249
  }
}

250
static int32_t fillResultImpl(SFillInfo* pFillInfo, SSDataBlock* pBlock, int32_t outputRows) {
251 252
  pFillInfo->numOfCurrent = 0;

253
  SColumnInfoData* pTsCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, pFillInfo->srcTsSlotId);
254 255

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

258 259 260
#if 0
  ASSERT(ascFill && (pFillInfo->currentKey >= pFillInfo->start) || (!ascFill && (pFillInfo->currentKey <= pFillInfo->start)));
#endif
261 262

  while (pFillInfo->numOfCurrent < outputRows) {
263
    int64_t ts = ((int64_t*)pTsCol->pData)[pFillInfo->index];
264 265

    // set the next value for interpolation
266
    if ((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) {
H
Haojun Liao 已提交
267
      copyCurrentRowIntoBuf(pFillInfo, pFillInfo->index, pFillInfo->next.pRowVal);
268 269
    }

X
Xiaoyu Wang 已提交
270 271
    if (((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) &&
        pFillInfo->numOfCurrent < outputRows) {
272
      // fill the gap between two input rows
X
Xiaoyu Wang 已提交
273 274
      while (((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) &&
             pFillInfo->numOfCurrent < outputRows) {
275
        doFillOneRow(pFillInfo, pBlock, pFillInfo->pSrcBlock, ts, false);
276 277 278 279 280 281 282 283
      }

      // output buffer is full, abort
      if (pFillInfo->numOfCurrent == outputRows) {
        pFillInfo->numOfTotal += pFillInfo->numOfCurrent;
        return outputRows;
      }
    } else {
284
      ASSERT(pFillInfo->currentKey == ts);
285
      int32_t index = pBlock->info.rows;
286

287
      if (pFillInfo->type == TSDB_FILL_NEXT && (pFillInfo->index + 1) < pFillInfo->numOfRows) {
288
        int32_t nextRowIndex = pFillInfo->index + 1;
H
Haojun Liao 已提交
289
        copyCurrentRowIntoBuf(pFillInfo, nextRowIndex, pFillInfo->next.pRowVal);
290 291
      }

H
Haojun Liao 已提交
292
      // copy rows to dst buffer
293 294 295
      for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
        SFillColInfo* pCol = &pFillInfo->pFillCol[i];

296
        int32_t dstSlotId = GET_DEST_SLOT_ID(pCol);
297

298
        SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, dstSlotId);
H
Haojun Liao 已提交
299
        SColumnInfoData* pSrc = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, dstSlotId);
300 301

        char* src = colDataGetData(pSrc, pFillInfo->index);
H
Haojun Liao 已提交
302 303
        if (!colDataIsNull_s(pSrc, pFillInfo->index)) {
          colDataAppend(pDst, index, src, false);
H
Haojun Liao 已提交
304
          saveColData(pFillInfo->prev.pRowVal, i, src, false);
H
Haojun Liao 已提交
305
        } else {  // the value is null
306 307 308 309
          if (pDst->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
            colDataAppend(pDst, index, (const char*)&pFillInfo->currentKey, false);
          } else {  // i > 0 and data is null , do interpolation
            if (pFillInfo->type == TSDB_FILL_PREV) {
H
Haojun Liao 已提交
310
              SArray*     p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->prev.pRowVal : pFillInfo->next.pRowVal;
311 312 313 314 315
              SGroupKeys* pKey = taosArrayGet(p, i);
              doSetVal(pDst, index, pKey);
            } else if (pFillInfo->type == TSDB_FILL_LINEAR) {
              bool isNull = colDataIsNull_s(pSrc, pFillInfo->index);
              colDataAppend(pDst, index, src, isNull);
H
Haojun Liao 已提交
316
              saveColData(pFillInfo->prev.pRowVal, i, src, isNull);  // todo:
317 318 319
            } else if (pFillInfo->type == TSDB_FILL_NULL) {
              colDataAppendNULL(pDst, index);
            } else if (pFillInfo->type == TSDB_FILL_NEXT) {
H
Haojun Liao 已提交
320
              SArray*     p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->next.pRowVal : pFillInfo->prev.pRowVal;
321 322 323 324 325 326
              SGroupKeys* pKey = taosArrayGet(p, i);
              doSetVal(pDst, index, pKey);
            } else {
              SVariant* pVar = &pFillInfo->pFillCol[i].fillVal;
              doSetUserSpecifiedValue(pDst, pVar, index, pFillInfo->currentKey);
            }
327 328 329 330 331
          }
        }
      }

      // set the tag value for final result
X
Xiaoyu Wang 已提交
332 333 334 335
      //      setTagsValue(pFillInfo, data, pFillInfo->numOfCurrent);
      SInterval* pInterval = &pFillInfo->interval;
      pFillInfo->currentKey =
          taosTimeAdd(pFillInfo->currentKey, pInterval->sliding * step, pInterval->slidingUnit, pInterval->precision);
336

337
      pBlock->info.rows += 1;
338 339 340 341 342 343 344 345 346 347 348 349 350
      pFillInfo->index += 1;
      pFillInfo->numOfCurrent += 1;
    }

    if (pFillInfo->index >= pFillInfo->numOfRows || pFillInfo->numOfCurrent >= outputRows) {
      pFillInfo->numOfTotal += pFillInfo->numOfCurrent;
      return pFillInfo->numOfCurrent;
    }
  }

  return pFillInfo->numOfCurrent;
}

351
static void saveColData(SArray* rowBuf, int32_t columnIndex, const char* src, bool isNull) {
X
Xiaoyu Wang 已提交
352
  SGroupKeys* pKey = taosArrayGet(rowBuf, columnIndex);
353 354 355
  if (isNull) {
    pKey->isNull = true;
  } else {
H
Haojun Liao 已提交
356 357 358 359 360
    if (IS_VAR_DATA_TYPE(pKey->type)) {
      memcpy(pKey->pData, src, varDataTLen(src));
    } else {
      memcpy(pKey->pData, src, pKey->bytes);
    }
361 362 363 364 365
    pKey->isNull = false;
  }
}

static int64_t appendFilledResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, int64_t resultCapacity) {
366 367 368 369 370 371
  /*
   * 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) {
372
    doFillOneRow(pFillInfo, pBlock, pFillInfo->pSrcBlock, pFillInfo->start, true);
373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388
  }

  pFillInfo->numOfTotal += pFillInfo->numOfCurrent;

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

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

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

H
Haojun Liao 已提交
389
struct SFillInfo* taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t numOfNotFillCols, int32_t capacity,
390
                                     SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol,
391
                                     int32_t primaryTsSlotId, int32_t order, const char* id) {
392 393 394 395
  if (fillType == TSDB_FILL_NONE) {
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
396
  SFillInfo* pFillInfo = taosMemoryCalloc(1, sizeof(SFillInfo));
397 398 399 400 401
  if (pFillInfo == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

402
  pFillInfo->order = order;
403 404 405 406
  pFillInfo->srcTsSlotId = primaryTsSlotId;

  for(int32_t i = 0; i < numOfNotFillCols; ++i) {
    SFillColInfo* p = &pCol[i + numOfFillCols];
H
Haojun Liao 已提交
407
    int32_t srcSlotId = GET_DEST_SLOT_ID(p);
408 409 410 411 412 413
    if (srcSlotId == primaryTsSlotId) {
      pFillInfo->tsSlotId = i + numOfFillCols;
      break;
    }
  }

414
  taosResetFillInfo(pFillInfo, skey);
415

X
Xiaoyu Wang 已提交
416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434
  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;
435 436 437 438 439
    default:
      terrno = TSDB_CODE_INVALID_PARA;
      return NULL;
  }

X
Xiaoyu Wang 已提交
440 441
  pFillInfo->type = fillType;
  pFillInfo->pFillCol = pCol;
H
Haojun Liao 已提交
442
  pFillInfo->numOfCols = numOfFillCols + numOfNotFillCols;
X
Xiaoyu Wang 已提交
443 444 445 446
  pFillInfo->alloc = capacity;
  pFillInfo->id = id;
  pFillInfo->interval = *pInterval;

H
Haojun Liao 已提交
447 448
  pFillInfo->next.pRowVal = taosArrayInit(pFillInfo->numOfCols, sizeof(SGroupKeys));
  pFillInfo->prev.pRowVal = taosArrayInit(pFillInfo->numOfCols, sizeof(SGroupKeys));
449 450

  initBeforeAfterDataBuf(pFillInfo);
451 452 453 454
  return pFillInfo;
}

void taosResetFillInfo(SFillInfo* pFillInfo, TSKEY startTimestamp) {
X
Xiaoyu Wang 已提交
455 456 457 458 459
  pFillInfo->start = startTimestamp;
  pFillInfo->currentKey = startTimestamp;
  pFillInfo->end = startTimestamp;
  pFillInfo->index = -1;
  pFillInfo->numOfRows = 0;
460
  pFillInfo->numOfCurrent = 0;
X
Xiaoyu Wang 已提交
461
  pFillInfo->numOfTotal = 0;
462 463 464 465 466 467
}

void* taosDestroyFillInfo(SFillInfo* pFillInfo) {
  if (pFillInfo == NULL) {
    return NULL;
  }
H
Haojun Liao 已提交
468 469
  for (int32_t i = 0; i < taosArrayGetSize(pFillInfo->prev.pRowVal); ++i) {
    SGroupKeys* pKey = taosArrayGet(pFillInfo->prev.pRowVal, i);
470 471
    taosMemoryFree(pKey->pData);
  }
H
Haojun Liao 已提交
472 473 474
  taosArrayDestroy(pFillInfo->prev.pRowVal);
  for (int32_t i = 0; i < taosArrayGetSize(pFillInfo->next.pRowVal); ++i) {
    SGroupKeys* pKey = taosArrayGet(pFillInfo->next.pRowVal, i);
475 476
    taosMemoryFree(pKey->pData);
  }
H
Haojun Liao 已提交
477
  taosArrayDestroy(pFillInfo->next.pRowVal);
478

H
Haojun Liao 已提交
479 480 481
//  for (int32_t i = 0; i < pFillInfo->numOfTags; ++i) {
//    taosMemoryFreeClear(pFillInfo->pTags[i].tagVal);
//  }
482

wafwerar's avatar
wafwerar 已提交
483 484 485
  taosMemoryFreeClear(pFillInfo->pTags);
  taosMemoryFreeClear(pFillInfo->pFillCol);
  taosMemoryFreeClear(pFillInfo);
486 487 488
  return NULL;
}

489 490 491 492 493 494 495 496
void taosFillSetDataOrderInfo(SFillInfo* pFillInfo, int32_t order) {
  if (pFillInfo == NULL || (order != TSDB_ORDER_ASC && order != TSDB_ORDER_DESC)) {
    return;
  }

  pFillInfo->order = order;
}

497 498 499 500 501 502 503
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)) {
504
    pFillInfo->end = taosTimeTruncate(endKey, &pFillInfo->interval, pFillInfo->interval.precision);
505 506
  }

X
Xiaoyu Wang 已提交
507
  pFillInfo->index = 0;
508 509 510 511
  pFillInfo->numOfRows = numOfRows;
}

void taosFillSetInputDataBlock(SFillInfo* pFillInfo, const SSDataBlock* pInput) {
X
Xiaoyu Wang 已提交
512
  pFillInfo->pSrcBlock = (SSDataBlock*)pInput;
513 514 515 516 517 518 519 520
}

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

521 522 523
  bool ascFill = FILL_IS_ASC_FILL(pFillInfo);
  if (pFillInfo->numOfTotal > 0 &&
      (((pFillInfo->end > pFillInfo->start) && ascFill) || (pFillInfo->end < pFillInfo->start && !ascFill))) {
524 525 526 527 528 529 530
    return getNumOfResultsAfterFillGap(pFillInfo, pFillInfo->end, 4096) > 0;
  }

  return false;
}

int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, TSKEY ekey, int32_t maxNumOfRows) {
531
  SColumnInfoData* pCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, pFillInfo->srcTsSlotId);
532

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

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

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

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

X
Xiaoyu Wang 已提交
561 562
int32_t taosGetLinearInterpolationVal(SPoint* point, int32_t outputType, SPoint* point1, SPoint* point2,
                                      int32_t inputType) {
563 564 565 566 567 568 569 570 571 572
  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;
}

573
int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, SSDataBlock* p, int32_t capacity) {
574 575 576 577 578 579 580
  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) {
581
    appendFilledResult(pFillInfo, p, numOfRes);
582
  } else {
X
Xiaoyu Wang 已提交
583
    fillResultImpl(pFillInfo, p, (int32_t)numOfRes);
584 585 586
    assert(numOfRes == pFillInfo->numOfCurrent);
  }

587
  qDebug("fill:%p, generated fill result, src block:%d, index:%d, brange:%" PRId64 "-%" PRId64 ", currentKey:%" PRId64
588 589 590
         ", current : % d, total : % d, %s",
         pFillInfo, pFillInfo->numOfRows, pFillInfo->index, pFillInfo->start, pFillInfo->end, pFillInfo->currentKey,
         pFillInfo->numOfCurrent, pFillInfo->numOfTotal, pFillInfo->id);
591 592 593

  return numOfRes;
}
594

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

H
Haojun Liao 已提交
597 598 599
SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprInfo* pNotFillExpr,
                                int32_t numOfNotFillExpr, const struct SNodeListNode* pValNode) {
  SFillColInfo* pFillCol = taosMemoryCalloc(numOfFillExpr + numOfNotFillExpr, sizeof(SFillColInfo));
600 601 602 603
  if (pFillCol == NULL) {
    return NULL;
  }

X
Xiaoyu Wang 已提交
604
  size_t len = (pValNode != NULL) ? LIST_LENGTH(pValNode->pNodeList) : 0;
H
Haojun Liao 已提交
605
  for (int32_t i = 0; i < numOfFillExpr; ++i) {
606 607
    SExprInfo* pExprInfo = &pExpr[i];
    pFillCol[i].pExpr = pExprInfo;
H
Haojun Liao 已提交
608
    pFillCol[i].notFillCol = false;
609

610 611 612
    // 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 已提交
613
      int32_t index = (i >= len) ? (len - 1) : i;
614 615

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

H
Haojun Liao 已提交
620 621 622 623 624 625
  for(int32_t i = 0; i < numOfNotFillExpr; ++i) {
    SExprInfo* pExprInfo = &pNotFillExpr[i];
    pFillCol[i + numOfFillExpr].pExpr = pExprInfo;
    pFillCol[i + numOfFillExpr].notFillCol = true;
  }

626
  return pFillCol;
627
}