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

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

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

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

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

H
Haojun Liao 已提交
38 39
static void doSetVal(SColumnInfoData* pDstColInfoData, int32_t rowIndex, const SGroupKeys* pKey);

D
dapan1121 已提交
40 41
static void setNotFillColumn(SFillInfo* pFillInfo, SColumnInfoData* pDstColInfo, int32_t rowIndex, int32_t colIdx) {
  SRowVal* p = NULL;
5
54liuyao 已提交
42 43
  if (pFillInfo->type == TSDB_FILL_NEXT) {
    p = FILL_IS_ASC_FILL(pFillInfo) ? &pFillInfo->next : &pFillInfo->prev;
D
dapan1121 已提交
44
  } else {
5
54liuyao 已提交
45
    p = FILL_IS_ASC_FILL(pFillInfo) ? &pFillInfo->prev : &pFillInfo->next;
D
dapan1121 已提交
46
  }
47

D
dapan1121 已提交
48 49 50 51
  SGroupKeys* pKey = taosArrayGet(p->pRowVal, colIdx);
  doSetVal(pDstColInfo, rowIndex, pKey);
}

H
Haojun Liao 已提交
52
static void setNullRow(SSDataBlock* pBlock, SFillInfo* pFillInfo, int32_t rowIndex) {
5
54liuyao 已提交
53 54 55
  for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
    SFillColInfo*    pCol = &pFillInfo->pFillCol[i];
    int32_t          dstSlotId = GET_DEST_SLOT_ID(pCol);
H
Haojun Liao 已提交
56 57
    SColumnInfoData* pDstColInfo = taosArrayGet(pBlock->pDataBlock, dstSlotId);
    if (pCol->notFillCol) {
58 59
      bool filled = fillIfWindowPseudoColumn(pFillInfo, pCol, pDstColInfo, rowIndex);
      if (!filled) {
D
dapan1121 已提交
60
        setNotFillColumn(pFillInfo, pDstColInfo, rowIndex, i);
H
Haojun Liao 已提交
61
      }
62
    } else {
63
      colDataSetNULL(pDstColInfo, rowIndex);
64
    }
65 66 67
  }
}

68
static void doSetUserSpecifiedValue(SColumnInfoData* pDst, SVariant* pVar, int32_t rowIndex, int64_t currentKey) {
69
  bool isNull = (TSDB_DATA_TYPE_NULL == pVar->nType) ? true : false;
70 71
  if (pDst->info.type == TSDB_DATA_TYPE_FLOAT) {
    float v = 0;
72 73
    GET_TYPED_DATA(v, float, pVar->nType, &pVar->f);
    colDataSetVal(pDst, rowIndex, (char*)&v, isNull);
74 75
  } else if (pDst->info.type == TSDB_DATA_TYPE_DOUBLE) {
    double v = 0;
76 77
    GET_TYPED_DATA(v, double, pVar->nType, &pVar->d);
    colDataSetVal(pDst, rowIndex, (char*)&v, isNull);
78 79 80
  } else if (IS_SIGNED_NUMERIC_TYPE(pDst->info.type)) {
    int64_t v = 0;
    GET_TYPED_DATA(v, int64_t, pVar->nType, &pVar->i);
81 82 83 84 85
    colDataSetVal(pDst, rowIndex, (char*)&v, isNull);
  } else if (IS_UNSIGNED_NUMERIC_TYPE(pDst->info.type)) {
    uint64_t v = 0;
    GET_TYPED_DATA(v, uint64_t, pVar->nType, &pVar->u);
    colDataSetVal(pDst, rowIndex, (char*)&v, isNull);
86
  } else if (pDst->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
87
    colDataSetVal(pDst, rowIndex, (const char*)&currentKey, isNull);
88
  } else {  // varchar/nchar data
89
    colDataSetNULL(pDst, rowIndex);
90 91 92
  }
}

5
54liuyao 已提交
93
// fill windows pseudo column, _wstart, _wend, _wduration and return true, otherwise return false
H
Haojun Liao 已提交
94 95
bool fillIfWindowPseudoColumn(SFillInfo* pFillInfo, SFillColInfo* pCol, SColumnInfoData* pDstColInfoData,
                              int32_t rowIndex) {
96 97 98 99 100 101 102 103
  if (!pCol->notFillCol) {
    return false;
  }
  if (pCol->pExpr->pExpr->nodeType == QUERY_NODE_COLUMN) {
    if (pCol->pExpr->base.numOfParams != 1) {
      return false;
    }
    if (pCol->pExpr->base.pParam[0].pCol->colType == COLUMN_TYPE_WINDOW_START) {
104
      colDataSetVal(pDstColInfoData, rowIndex, (const char*)&pFillInfo->currentKey, false);
105 106
      return true;
    } else if (pCol->pExpr->base.pParam[0].pCol->colType == COLUMN_TYPE_WINDOW_END) {
5
54liuyao 已提交
107
      // TODO: include endpoint
108
      SInterval* pInterval = &pFillInfo->interval;
5
54liuyao 已提交
109
      int64_t    windowEnd =
S
shenglian zhou 已提交
110
          taosTimeAdd(pFillInfo->currentKey, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
111
      colDataSetVal(pDstColInfoData, rowIndex, (const char*)&windowEnd, false);
112 113
      return true;
    } else if (pCol->pExpr->base.pParam[0].pCol->colType == COLUMN_TYPE_WINDOW_DURATION) {
5
54liuyao 已提交
114
      // TODO: include endpoint
115
      colDataSetVal(pDstColInfoData, rowIndex, (const char*)&pFillInfo->interval.sliding, false);
116 117 118
      return true;
    }
  }
119 120 121
  return false;
}

122 123
static void doFillOneRow(SFillInfo* pFillInfo, SSDataBlock* pBlock, SSDataBlock* pSrcBlock, int64_t ts,
                         bool outOfBound) {
X
Xiaoyu Wang 已提交
124
  SPoint  point1, point2, point;
125 126
  int32_t step = GET_FORWARD_DIRECTION_FACTOR(pFillInfo->order);

127
  // set the primary timestamp column value
128
  int32_t index = pBlock->info.rows;
129 130 131

  // set the other values
  if (pFillInfo->type == TSDB_FILL_PREV) {
132
    for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
133 134 135
      SFillColInfo* pCol = &pFillInfo->pFillCol[i];

      SColumnInfoData* pDstColInfoData = taosArrayGet(pBlock->pDataBlock, GET_DEST_SLOT_ID(pCol));
5
54liuyao 已提交
136
      bool             filled = fillIfWindowPseudoColumn(pFillInfo, pCol, pDstColInfoData, index);
137
      if (!filled) {
D
dapan1121 已提交
138
        setNotFillColumn(pFillInfo, pDstColInfoData, index, i);
139
      }
140 141
    }
  } else if (pFillInfo->type == TSDB_FILL_NEXT) {
142
    // todo  refactor: start from 0 not 1
143
    for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
5
54liuyao 已提交
144
      SFillColInfo*    pCol = &pFillInfo->pFillCol[i];
145
      SColumnInfoData* pDstColInfoData = taosArrayGet(pBlock->pDataBlock, GET_DEST_SLOT_ID(pCol));
5
54liuyao 已提交
146
      bool             filled = fillIfWindowPseudoColumn(pFillInfo, pCol, pDstColInfoData, index);
147
      if (!filled) {
D
dapan1121 已提交
148
        setNotFillColumn(pFillInfo, pDstColInfoData, index, i);
149
      }
150 151 152
    }
  } else if (pFillInfo->type == TSDB_FILL_LINEAR) {
    // TODO : linear interpolation supports NULL value
153
    if (outOfBound) {
H
Haojun Liao 已提交
154
      setNullRow(pBlock, pFillInfo, index);
155
    } else {
156
      for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
157 158
        SFillColInfo* pCol = &pFillInfo->pFillCol[i];

X
Xiaoyu Wang 已提交
159
        int32_t          dstSlotId = GET_DEST_SLOT_ID(pCol);
160
        SColumnInfoData* pDstCol = taosArrayGet(pBlock->pDataBlock, dstSlotId);
H
Haojun Liao 已提交
161 162 163
        int16_t          type = pDstCol->info.type;

        if (pCol->notFillCol) {
164 165
          bool filled = fillIfWindowPseudoColumn(pFillInfo, pCol, pDstCol, index);
          if (!filled) {
D
dapan1121 已提交
166
            setNotFillColumn(pFillInfo, pDstCol, index, i);
H
Haojun Liao 已提交
167 168
          }
        } else {
H
Haojun Liao 已提交
169
          SGroupKeys* pKey = taosArrayGet(pFillInfo->prev.pRowVal, i);
H
Haojun Liao 已提交
170
          if (IS_VAR_DATA_TYPE(type) || type == TSDB_DATA_TYPE_BOOL || pKey->isNull) {
171
            colDataSetNULL(pDstCol, index);
H
Haojun Liao 已提交
172 173
            continue;
          }
174

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

H
Haojun Liao 已提交
177
          int64_t prevTs = *(int64_t*)pKey1->pData;
H
Haojun Liao 已提交
178
          int32_t srcSlotId = GET_DEST_SLOT_ID(pCol);
179

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

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

H
Haojun Liao 已提交
186 187 188
          int64_t out = 0;
          point = (SPoint){.key = pFillInfo->currentKey, .val = &out};
          taosGetLinearInterpolationVal(&point, type, &point1, &point2, type);
189

190
          colDataSetVal(pDstCol, index, (const char*)&out, false);
H
Haojun Liao 已提交
191
        }
192 193
      }
    }
D
dapan1121 已提交
194
  } else if (pFillInfo->type == TSDB_FILL_NULL || pFillInfo->type == TSDB_FILL_NULL_F) {  // fill with NULL
H
Haojun Liao 已提交
195
    setNullRow(pBlock, pFillInfo, index);
X
Xiaoyu Wang 已提交
196
  } else {  // fill with user specified value for each column
197
    for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
198 199
      SFillColInfo* pCol = &pFillInfo->pFillCol[i];

5
54liuyao 已提交
200
      int32_t          slotId = GET_DEST_SLOT_ID(pCol);
H
Haojun Liao 已提交
201 202 203
      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, slotId);

      if (pCol->notFillCol) {
204 205
        bool filled = fillIfWindowPseudoColumn(pFillInfo, pCol, pDst, index);
        if (!filled) {
D
dapan1121 已提交
206
          setNotFillColumn(pFillInfo, pDst, index, i);
H
Haojun Liao 已提交
207 208 209 210 211
        }
      } else {
        SVariant* pVar = &pFillInfo->pFillCol[i].fillVal;
        doSetUserSpecifiedValue(pDst, pVar, index, pFillInfo->currentKey);
      }
212 213 214
    }
  }

X
Xiaoyu Wang 已提交
215
  //  setTagsValue(pFillInfo, data, index);
216
  SInterval* pInterval = &pFillInfo->interval;
X
Xiaoyu Wang 已提交
217 218
  pFillInfo->currentKey =
      taosTimeAdd(pFillInfo->currentKey, pInterval->sliding * step, pInterval->slidingUnit, pInterval->precision);
219
  pBlock->info.rows += 1;
220 221 222
  pFillInfo->numOfCurrent++;
}

223 224
void doSetVal(SColumnInfoData* pDstCol, int32_t rowIndex, const SGroupKeys* pKey) {
  if (pKey->isNull) {
225
    colDataSetNULL(pDstCol, rowIndex);
226
  } else {
227
    colDataSetVal(pDstCol, rowIndex, pKey->pData, false);
228 229 230 231
  }
}

static void initBeforeAfterDataBuf(SFillInfo* pFillInfo) {
H
Haojun Liao 已提交
232
  if (taosArrayGetSize(pFillInfo->next.pRowVal) > 0) {
233 234 235
    return;
  }

236
  for (int i = 0; i < pFillInfo->numOfCols; i++) {
237
    SFillColInfo* pCol = &pFillInfo->pFillCol[i];
238

X
Xiaoyu Wang 已提交
239
    SGroupKeys  key = {0};
240
    SResSchema* pSchema = &pCol->pExpr->base.resSchema;
X
Xiaoyu Wang 已提交
241
    key.pData = taosMemoryMalloc(pSchema->bytes);
242 243
    key.isNull = true;
    key.bytes = pSchema->bytes;
X
Xiaoyu Wang 已提交
244
    key.type = pSchema->type;
245

H
Haojun Liao 已提交
246
    taosArrayPush(pFillInfo->next.pRowVal, &key);
247 248

    key.pData = taosMemoryMalloc(pSchema->bytes);
H
Haojun Liao 已提交
249
    taosArrayPush(pFillInfo->prev.pRowVal, &key);
250 251 252
  }
}

253 254
static void saveColData(SArray* rowBuf, int32_t columnIndex, const char* src, bool isNull);

255
static void copyCurrentRowIntoBuf(SFillInfo* pFillInfo, int32_t rowIndex, SRowVal* pRowVal, bool reset) {
256 257 258
  SColumnInfoData* pTsCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, pFillInfo->srcTsSlotId);
  pRowVal->key = ((int64_t*)pTsCol->pData)[rowIndex];

259
  for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
H
Haojun Liao 已提交
260
    int32_t type = pFillInfo->pFillCol[i].pExpr->pExpr->nodeType;
5
54liuyao 已提交
261 262 263 264
    if ( type == QUERY_NODE_COLUMN || type == QUERY_NODE_OPERATOR || type == QUERY_NODE_FUNCTION) {
      if (!pFillInfo->pFillCol[i].notFillCol && pFillInfo->type != TSDB_FILL_NEXT) {
        continue;
      }
H
Haojun Liao 已提交
265
      int32_t srcSlotId = GET_DEST_SLOT_ID(&pFillInfo->pFillCol[i]);
266

5
54liuyao 已提交
267 268 269 270
      if (srcSlotId == pFillInfo->srcTsSlotId && pFillInfo->type == TSDB_FILL_LINEAR) {
        continue;
      }

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

H
Haojun Liao 已提交
273 274
      bool  isNull = colDataIsNull_s(pSrcCol, rowIndex);
      char* p = colDataGetData(pSrcCol, rowIndex);
275

276
      saveColData(pRowVal->pRowVal, i, p, reset ? true : isNull);
H
Haojun Liao 已提交
277 278 279
    } else {
      ASSERT(0);
    }
280 281 282
  }
}

283
static int32_t fillResultImpl(SFillInfo* pFillInfo, SSDataBlock* pBlock, int32_t outputRows) {
284 285
  pFillInfo->numOfCurrent = 0;

286
  SColumnInfoData* pTsCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, pFillInfo->srcTsSlotId);
287 288

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

291 292 293
#if 0
  ASSERT(ascFill && (pFillInfo->currentKey >= pFillInfo->start) || (!ascFill && (pFillInfo->currentKey <= pFillInfo->start)));
#endif
294 295

  while (pFillInfo->numOfCurrent < outputRows) {
296
    int64_t ts = ((int64_t*)pTsCol->pData)[pFillInfo->index];
297 298

    // set the next value for interpolation
5
54liuyao 已提交
299 300
    if (pFillInfo->currentKey < ts && ascFill) {
      SRowVal* pRVal = pFillInfo->type == TSDB_FILL_NEXT ? &pFillInfo->next : &pFillInfo->prev;
301
      copyCurrentRowIntoBuf(pFillInfo, pFillInfo->index, pRVal, false);
5
54liuyao 已提交
302 303
    } else if (pFillInfo->currentKey > ts && !ascFill) {
      SRowVal* pRVal = pFillInfo->type == TSDB_FILL_NEXT ? &pFillInfo->prev : &pFillInfo->next;
304
      copyCurrentRowIntoBuf(pFillInfo, pFillInfo->index, pRVal, false);
305 306
    }

X
Xiaoyu Wang 已提交
307 308
    if (((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) &&
        pFillInfo->numOfCurrent < outputRows) {
309
      // fill the gap between two input rows
X
Xiaoyu Wang 已提交
310 311
      while (((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) &&
             pFillInfo->numOfCurrent < outputRows) {
312
        doFillOneRow(pFillInfo, pBlock, pFillInfo->pSrcBlock, ts, false);
313 314 315 316 317 318 319 320
      }

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

324
      if (pFillInfo->type == TSDB_FILL_NEXT) {
325
        int32_t nextRowIndex = pFillInfo->index + 1;
326 327 328 329 330 331
        if ((pFillInfo->index + 1) < pFillInfo->numOfRows) {
          copyCurrentRowIntoBuf(pFillInfo, nextRowIndex, &pFillInfo->next, false);
        } else {
          // reset to null after last row
          copyCurrentRowIntoBuf(pFillInfo, nextRowIndex, &pFillInfo->next, true);
        }
332 333
      }

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

338
        int32_t dstSlotId = GET_DEST_SLOT_ID(pCol);
339

340
        SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, dstSlotId);
H
Haojun Liao 已提交
341
        SColumnInfoData* pSrc = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, dstSlotId);
342 343

        char* src = colDataGetData(pSrc, pFillInfo->index);
H
Haojun Liao 已提交
344
        if (!colDataIsNull_s(pSrc, pFillInfo->index)) {
345
          colDataSetVal(pDst, index, src, false);
H
Haojun Liao 已提交
346
          saveColData(pFillInfo->prev.pRowVal, i, src, false);
347 348 349
          if (pFillInfo->srcTsSlotId == dstSlotId) {
            pFillInfo->prev.key = *(int64_t*)src;
          }
H
Haojun Liao 已提交
350
        } else {  // the value is null
351
          if (pDst->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
352
            colDataSetVal(pDst, index, (const char*)&pFillInfo->currentKey, false);
353 354
          } else {  // i > 0 and data is null , do interpolation
            if (pFillInfo->type == TSDB_FILL_PREV) {
H
Haojun Liao 已提交
355
              SArray*     p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->prev.pRowVal : pFillInfo->next.pRowVal;
356 357 358 359
              SGroupKeys* pKey = taosArrayGet(p, i);
              doSetVal(pDst, index, pKey);
            } else if (pFillInfo->type == TSDB_FILL_LINEAR) {
              bool isNull = colDataIsNull_s(pSrc, pFillInfo->index);
360
              colDataSetVal(pDst, index, src, isNull);
H
Haojun Liao 已提交
361
              saveColData(pFillInfo->prev.pRowVal, i, src, isNull);  // todo:
D
dapan1121 已提交
362
            } else if (pFillInfo->type == TSDB_FILL_NULL || pFillInfo->type == TSDB_FILL_NULL_F) {
363
              colDataSetNULL(pDst, index);
364
            } else if (pFillInfo->type == TSDB_FILL_NEXT) {
H
Haojun Liao 已提交
365
              SArray*     p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->next.pRowVal : pFillInfo->prev.pRowVal;
366 367 368 369 370 371
              SGroupKeys* pKey = taosArrayGet(p, i);
              doSetVal(pDst, index, pKey);
            } else {
              SVariant* pVar = &pFillInfo->pFillCol[i].fillVal;
              doSetUserSpecifiedValue(pDst, pVar, index, pFillInfo->currentKey);
            }
372 373 374 375 376
          }
        }
      }

      // set the tag value for final result
X
Xiaoyu Wang 已提交
377 378 379
      SInterval* pInterval = &pFillInfo->interval;
      pFillInfo->currentKey =
          taosTimeAdd(pFillInfo->currentKey, pInterval->sliding * step, pInterval->slidingUnit, pInterval->precision);
380

381
      pBlock->info.rows += 1;
382 383 384 385 386 387 388 389 390 391 392 393 394
      pFillInfo->index += 1;
      pFillInfo->numOfCurrent += 1;
    }

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

  return pFillInfo->numOfCurrent;
}

395
static void saveColData(SArray* rowBuf, int32_t columnIndex, const char* src, bool isNull) {
X
Xiaoyu Wang 已提交
396
  SGroupKeys* pKey = taosArrayGet(rowBuf, columnIndex);
397 398 399
  if (isNull) {
    pKey->isNull = true;
  } else {
H
Haojun Liao 已提交
400 401 402 403 404
    if (IS_VAR_DATA_TYPE(pKey->type)) {
      memcpy(pKey->pData, src, varDataTLen(src));
    } else {
      memcpy(pKey->pData, src, pKey->bytes);
    }
405 406 407 408 409
    pKey->isNull = false;
  }
}

static int64_t appendFilledResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, int64_t resultCapacity) {
410 411 412 413 414 415
  /*
   * 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) {
416
    doFillOneRow(pFillInfo, pBlock, pFillInfo->pSrcBlock, pFillInfo->start, true);
417 418 419 420
  }

  pFillInfo->numOfTotal += pFillInfo->numOfCurrent;

H
Haojun Liao 已提交
421
  ASSERT(pFillInfo->numOfCurrent == resultCapacity);
422 423 424 425 426 427 428 429 430 431 432
  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 已提交
433
struct SFillInfo* taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t numOfNotFillCols, int32_t capacity,
434
                                     SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol,
435
                                     int32_t primaryTsSlotId, int32_t order, const char* id) {
436 437 438 439
  if (fillType == TSDB_FILL_NONE) {
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
440
  SFillInfo* pFillInfo = taosMemoryCalloc(1, sizeof(SFillInfo));
441 442 443 444 445
  if (pFillInfo == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

446
  pFillInfo->order = order;
447 448
  pFillInfo->srcTsSlotId = primaryTsSlotId;

5
54liuyao 已提交
449
  for (int32_t i = 0; i < numOfNotFillCols; ++i) {
450
    SFillColInfo* p = &pCol[i + numOfFillCols];
5
54liuyao 已提交
451
    int32_t       srcSlotId = GET_DEST_SLOT_ID(p);
452 453 454 455 456 457
    if (srcSlotId == primaryTsSlotId) {
      pFillInfo->tsSlotId = i + numOfFillCols;
      break;
    }
  }

458
  taosResetFillInfo(pFillInfo, skey);
459

X
Xiaoyu Wang 已提交
460 461
  pFillInfo->type = fillType;
  pFillInfo->pFillCol = pCol;
H
Haojun Liao 已提交
462
  pFillInfo->numOfCols = numOfFillCols + numOfNotFillCols;
X
Xiaoyu Wang 已提交
463 464 465 466
  pFillInfo->alloc = capacity;
  pFillInfo->id = id;
  pFillInfo->interval = *pInterval;

H
Haojun Liao 已提交
467 468
  pFillInfo->next.pRowVal = taosArrayInit(pFillInfo->numOfCols, sizeof(SGroupKeys));
  pFillInfo->prev.pRowVal = taosArrayInit(pFillInfo->numOfCols, sizeof(SGroupKeys));
469 470

  initBeforeAfterDataBuf(pFillInfo);
471 472 473 474
  return pFillInfo;
}

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

void* taosDestroyFillInfo(SFillInfo* pFillInfo) {
  if (pFillInfo == NULL) {
    return NULL;
  }
H
Haojun Liao 已提交
488 489
  for (int32_t i = 0; i < taosArrayGetSize(pFillInfo->prev.pRowVal); ++i) {
    SGroupKeys* pKey = taosArrayGet(pFillInfo->prev.pRowVal, i);
490 491
    taosMemoryFree(pKey->pData);
  }
H
Haojun Liao 已提交
492 493 494
  taosArrayDestroy(pFillInfo->prev.pRowVal);
  for (int32_t i = 0; i < taosArrayGetSize(pFillInfo->next.pRowVal); ++i) {
    SGroupKeys* pKey = taosArrayGet(pFillInfo->next.pRowVal, i);
495 496
    taosMemoryFree(pKey->pData);
  }
H
Haojun Liao 已提交
497
  taosArrayDestroy(pFillInfo->next.pRowVal);
498

5
54liuyao 已提交
499 500 501
  //  for (int32_t i = 0; i < pFillInfo->numOfTags; ++i) {
  //    taosMemoryFreeClear(pFillInfo->pTags[i].tagVal);
  //  }
502

503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518
  // free pFillCol
  if (pFillInfo->pFillCol) {
    for (int32_t i = 0; i < pFillInfo->numOfCols; i++) {
      SFillColInfo* pCol = &pFillInfo->pFillCol[i];
      if (!pCol->notFillCol) {
        if (pCol->fillVal.nType == TSDB_DATA_TYPE_VARBINARY || pCol->fillVal.nType == TSDB_DATA_TYPE_VARCHAR ||
            pCol->fillVal.nType == TSDB_DATA_TYPE_NCHAR || pCol->fillVal.nType == TSDB_DATA_TYPE_JSON) {
          if (pCol->fillVal.pz) {
            taosMemoryFree(pCol->fillVal.pz);
            pCol->fillVal.pz = NULL;
          }
        }
      }
    }
  }

wafwerar's avatar
wafwerar 已提交
519 520 521
  taosMemoryFreeClear(pFillInfo->pTags);
  taosMemoryFreeClear(pFillInfo->pFillCol);
  taosMemoryFreeClear(pFillInfo);
522 523 524 525 526 527 528 529
  return NULL;
}

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

530
  // the endKey is now the aligned time window value. truncate time window isn't correct.
531
  pFillInfo->end = endKey;
532 533 534 535 536 537

#if 0
  if (pFillInfo->order == TSDB_ORDER_ASC) {
    ASSERT(pFillInfo->start <= pFillInfo->end);
  } else {
    ASSERT(pFillInfo->start >= pFillInfo->end);
538
  }
539
#endif
540

X
Xiaoyu Wang 已提交
541
  pFillInfo->index = 0;
542 543 544 545
  pFillInfo->numOfRows = numOfRows;
}

void taosFillSetInputDataBlock(SFillInfo* pFillInfo, const SSDataBlock* pInput) {
X
Xiaoyu Wang 已提交
546
  pFillInfo->pSrcBlock = (SSDataBlock*)pInput;
547 548
}

549 550 551 552 553 554 555
void taosFillUpdateStartTimestampInfo(SFillInfo* pFillInfo, int64_t ts) {
  pFillInfo->start = ts;
  pFillInfo->currentKey = ts;
}

bool taosFillNotStarted(const SFillInfo* pFillInfo) {return pFillInfo->start == pFillInfo->currentKey;}

556 557 558 559 560 561
bool taosFillHasMoreResults(SFillInfo* pFillInfo) {
  int32_t remain = taosNumOfRemainRows(pFillInfo);
  if (remain > 0) {
    return true;
  }

562 563 564
  bool ascFill = FILL_IS_ASC_FILL(pFillInfo);
  if (pFillInfo->numOfTotal > 0 &&
      (((pFillInfo->end > pFillInfo->start) && ascFill) || (pFillInfo->end < pFillInfo->start && !ascFill))) {
565 566 567 568 569 570 571
    return getNumOfResultsAfterFillGap(pFillInfo, pFillInfo->end, 4096) > 0;
  }

  return false;
}

int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, TSKEY ekey, int32_t maxNumOfRows) {
X
Xiaoyu Wang 已提交
572
  int32_t  numOfRows = taosNumOfRemainRows(pFillInfo);
573 574

  TSKEY ekey1 = ekey;
H
Hongze Cheng 已提交
575

576 577
  int64_t numOfRes = -1;
  if (numOfRows > 0) {  // still fill gap within current data block, not generating data after the result set.
D
dapan1121 已提交
578 579
    SColumnInfoData* pCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, pFillInfo->srcTsSlotId);    
    int64_t* tsList = (int64_t*)pCol->pData;
S
slzhou 已提交
580
    TSKEY lastKey = tsList[pFillInfo->numOfRows - 1];
X
Xiaoyu Wang 已提交
581 582
    numOfRes = taosTimeCountInterval(lastKey, pFillInfo->currentKey, pFillInfo->interval.sliding,
                                     pFillInfo->interval.slidingUnit, pFillInfo->interval.precision);
583
    numOfRes += 1;
H
Haojun Liao 已提交
584
    ASSERT(numOfRes >= numOfRows);
X
Xiaoyu Wang 已提交
585
  } else {  // reach the end of data
586
    if ((ekey1 < pFillInfo->currentKey && FILL_IS_ASC_FILL(pFillInfo)) ||
587
        (ekey1 > pFillInfo->currentKey && !FILL_IS_ASC_FILL(pFillInfo))) {
588 589
      return 0;
    }
590

X
Xiaoyu Wang 已提交
591 592
    numOfRes = taosTimeCountInterval(ekey1, pFillInfo->currentKey, pFillInfo->interval.sliding,
                                     pFillInfo->interval.slidingUnit, pFillInfo->interval.precision);
593 594 595 596 597 598
    numOfRes += 1;
  }

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

X
Xiaoyu Wang 已提交
599 600
int32_t taosGetLinearInterpolationVal(SPoint* point, int32_t outputType, SPoint* point1, SPoint* point2,
                                      int32_t inputType) {
601 602 603 604
  double v1 = -1, v2 = -1;
  GET_TYPED_DATA(v1, double, inputType, point1->val);
  GET_TYPED_DATA(v2, double, inputType, point2->val);

605 606 607 608 609 610
  double r = 0;
  if (!IS_BOOLEAN_TYPE(inputType)) {
    r = DO_INTERPOLATION(v1, v2, point1->key, point2->key, point->key);
  } else {
    r = (v1 < 1 || v2 < 1) ? 0 : 1;
  }
611 612 613 614 615
  SET_TYPED_DATA(point->val, outputType, r);

  return TSDB_CODE_SUCCESS;
}

616
int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, SSDataBlock* p, int32_t capacity) {
617 618 619
  int32_t remain = taosNumOfRemainRows(pFillInfo);

  int64_t numOfRes = getNumOfResultsAfterFillGap(pFillInfo, pFillInfo->end, capacity);
H
Haojun Liao 已提交
620
  ASSERT(numOfRes <= capacity);
621 622 623

  // no data existed for fill operation now, append result according to the fill strategy
  if (remain == 0) {
624
    appendFilledResult(pFillInfo, p, numOfRes);
625
  } else {
X
Xiaoyu Wang 已提交
626
    fillResultImpl(pFillInfo, p, (int32_t)numOfRes);
H
Haojun Liao 已提交
627
    ASSERT(numOfRes == pFillInfo->numOfCurrent);
628 629
  }

630
  qDebug("fill:%p, generated fill result, src block:%d, index:%d, brange:%" PRId64 "-%" PRId64 ", currentKey:%" PRId64
631 632 633
         ", current : % d, total : % d, %s",
         pFillInfo, pFillInfo->numOfRows, pFillInfo->index, pFillInfo->start, pFillInfo->end, pFillInfo->currentKey,
         pFillInfo->numOfCurrent, pFillInfo->numOfTotal, pFillInfo->id);
634 635 636

  return numOfRes;
}
637

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

H
Haojun Liao 已提交
640
SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprInfo* pNotFillExpr,
641 642
                                int32_t numOfNoFillExpr, const struct SNodeListNode* pValNode) {
  SFillColInfo* pFillCol = taosMemoryCalloc(numOfFillExpr + numOfNoFillExpr, sizeof(SFillColInfo));
643 644 645 646
  if (pFillCol == NULL) {
    return NULL;
  }

X
Xiaoyu Wang 已提交
647
  size_t len = (pValNode != NULL) ? LIST_LENGTH(pValNode->pNodeList) : 0;
H
Haojun Liao 已提交
648
  for (int32_t i = 0; i < numOfFillExpr; ++i) {
649 650
    SExprInfo* pExprInfo = &pExpr[i];
    pFillCol[i].pExpr = pExprInfo;
H
Haojun Liao 已提交
651
    pFillCol[i].notFillCol = false;
652

653 654 655
    // 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 已提交
656
      int32_t index = (i >= len) ? (len - 1) : i;
657 658

      SValueNode* pv = (SValueNode*)nodesListGetNode(pValNode->pNodeList, index);
X
Xiaoyu Wang 已提交
659
      nodesValueNodeToVariant(pv, &pFillCol[i].fillVal);
660
    }
661
  }
G
Ganlin Zhao 已提交
662
  pFillCol->numOfFillExpr = numOfFillExpr;
663

664
  for (int32_t i = 0; i < numOfNoFillExpr; ++i) {
H
Haojun Liao 已提交
665 666 667 668 669
    SExprInfo* pExprInfo = &pNotFillExpr[i];
    pFillCol[i + numOfFillExpr].pExpr = pExprInfo;
    pFillCol[i + numOfFillExpr].notFillCol = true;
  }

670
  return pFillCol;
671
}