builtinsimpl.c 88.2 KB
Newer Older
H
Haojun Liao 已提交
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 "builtinsimpl.h"
17
#include "cJSON.h"
18
#include "function.h"
19
#include "querynodes.h"
H
Haojun Liao 已提交
20 21
#include "taggfunction.h"
#include "tdatablock.h"
22
#include "tpercentile.h"
H
Haojun Liao 已提交
23

24
#define HISTOGRAM_MAX_BINS_NUM   100
25

G
Ganlin Zhao 已提交
26 27 28 29 30 31 32 33 34 35 36 37 38 39
typedef struct SSumRes {
  union {
    int64_t  isum;
    uint64_t usum;
    double   dsum;
  };
} SSumRes;

typedef struct SAvgRes {
  double  result;
  SSumRes sum;
  int64_t count;
} SAvgRes;

40 41 42 43 44
typedef struct STuplePos {
 int32_t pageId;
 int32_t offset;
} STuplePos;

45
typedef struct STopBotResItem {
46 47 48
  SVariant  v;
  uint64_t  uid;  // it is a table uid, used to extract tag data during building of the final result for the tag data
  STuplePos tuplePos;  // tuple data of this chosen row
49 50
} STopBotResItem;

G
Ganlin Zhao 已提交
51
typedef struct STopBotRes {
52
  STopBotResItem* pItems;
G
Ganlin Zhao 已提交
53 54 55 56 57
} STopBotRes;

typedef struct SStddevRes {
  double  result;
  int64_t count;
58 59 60 61 62 63 64 65
  union {
    double  quadraticDSum;
    int64_t quadraticISum;
  };
  union {
    double  dsum;
    int64_t isum;
  };
G
Ganlin Zhao 已提交
66 67
} SStddevRes;

68 69 70 71 72 73 74
typedef struct SLeastSQRInfo {
  double matrix[2][3];
  double startVal;
  double stepVal;
  int64_t num;
} SLeastSQRInfo;

G
Ganlin Zhao 已提交
75 76
typedef struct SPercentileInfo {
  double      result;
77
  tMemBucket* pMemBucket;
G
Ganlin Zhao 已提交
78 79 80 81 82 83 84
  int32_t     stage;
  double      minval;
  double      maxval;
  int64_t     numOfElems;
} SPercentileInfo;

typedef struct SDiffInfo {
85 86
  bool hasPrev;
  bool includeNull;
87
  bool ignoreNegative;  // replace the ignore with case when
88 89 90 91 92
  bool firstOutput;
  union {
    int64_t i64;
    double  d64;
  } prev;
93 94

  int64_t prevTs;
G
Ganlin Zhao 已提交
95 96
} SDiffInfo;

G
Ganlin Zhao 已提交
97 98 99 100 101 102 103
typedef struct SSpreadInfo {
  double result;
  bool   hasResult;
  double min;
  double max;
} SSpreadInfo;

104 105 106 107 108 109 110 111 112 113 114
typedef struct SHistoFuncBin {
  double lower;
  double upper;
  union {
    int64_t count;
    double  percentage;
  };
} SHistoFuncBin;

typedef struct SHistoFuncInfo {
  int32_t numOfBins;
115
  int32_t totalCount;
116 117 118 119
  bool    normalized;
  SHistoFuncBin bins[];
} SHistoFuncInfo;

120 121 122 123 124 125 126
typedef enum {
  UNKNOWN_BIN = 0,
  USER_INPUT_BIN,
  LINEAR_BIN,
  LOG_BIN
} EHistoBinType;

127 128 129 130 131 132 133 134 135 136 137 138 139
typedef struct SStateInfo {
  int64_t count;
} SStateInfo;

typedef enum {
  STATE_OPER_INVALID = 0,
  STATE_OPER_LT,
  STATE_OPER_GT,
  STATE_OPER_LE,
  STATE_OPER_GE,
  STATE_OPER_NE,
  STATE_OPER_EQ,
} EStateOperType;
140

141 142 143 144 145 146
#define SET_VAL(_info, numOfElem, res) \
  do {                                 \
    if ((numOfElem) <= 0) {            \
      break;                           \
    }                                  \
    (_info)->numOfRes = (res);         \
H
Haojun Liao 已提交
147 148
  } while (0)

G
Ganlin Zhao 已提交
149 150 151 152 153 154
#define GET_TS_LIST(x)    ((TSKEY*)((x)->ptsList))
#define GET_TS_DATA(x, y) (GET_TS_LIST(x)[(y)])

#define DO_UPDATE_TAG_COLUMNS_WITHOUT_TS(ctx)                      \
  do {                                                             \
    for (int32_t _i = 0; _i < (ctx)->tagInfo.numOfTagCols; ++_i) { \
155
      SqlFunctionCtx* __ctx = (ctx)->tagInfo.pTagCtxList[_i];      \
G
Ganlin Zhao 已提交
156 157 158 159 160 161 162 163 164 165 166 167 168 169 170
      __ctx->fpSet.process(__ctx);                                 \
    }                                                              \
  } while (0);

#define UPDATE_DATA(ctx, left, right, num, sign, _ts) \
  do {                                                \
    if (((left) < (right)) ^ (sign)) {                \
      (left) = (right);                               \
      DO_UPDATE_SUBSID_RES(ctx, _ts);                 \
      (num) += 1;                                     \
    }                                                 \
  } while (0)

#define LOOPCHECK_N(val, _col, ctx, _t, _nrow, _start, sign, num)        \
  do {                                                                   \
171
    _t* d = (_t*)((_col)->pData);                                        \
G
Ganlin Zhao 已提交
172 173 174 175 176 177 178 179 180
    for (int32_t i = (_start); i < (_nrow) + (_start); ++i) {            \
      if (((_col)->hasNull) && colDataIsNull_f((_col)->nullbitmap, i)) { \
        continue;                                                        \
      }                                                                  \
      TSKEY ts = (ctx)->ptsList != NULL ? GET_TS_DATA(ctx, i) : 0;       \
      UPDATE_DATA(ctx, val, d[i], num, sign, ts);                        \
    }                                                                    \
  } while (0)

181
bool functionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {
H
Haojun Liao 已提交
182 183 184 185 186 187 188 189 190 191 192 193
  if (pResultInfo->initialized) {
    return false;
  }

  if (pCtx->pOutput != NULL) {
    memset(pCtx->pOutput, 0, (size_t)pCtx->resDataInfo.bytes);
  }

  initResultRowEntry(pResultInfo, pCtx->resDataInfo.interBufSize);
  return true;
}

194
int32_t functionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
195
  int32_t          slotId = pCtx->pExpr->base.resSchema.slotId;
196
  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
197

198
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
199 200
  pResInfo->isNullRes = (pResInfo->numOfRes == 0) ? 1 : 0;
  /*cleanupResultRowEntry(pResInfo);*/
201 202 203 204 205

  char* in = GET_ROWCELL_INTERBUF(pResInfo);
  colDataAppend(pCol, pBlock->info.rows, in, pResInfo->isNullRes);

  return pResInfo->numOfRes;
H
Haojun Liao 已提交
206 207
}

208
int32_t functionFinalizeWithResultBuf(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, char* finalResult) {
209
  int32_t          slotId = pCtx->pExpr->base.resSchema.slotId;
210 211 212
  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);

  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
213
  pResInfo->isNullRes = (pResInfo->numOfRes == 0) ? 1 : 0;
214 215 216 217 218 219 220 221
  cleanupResultRowEntry(pResInfo);

  char* in = finalResult;
  colDataAppend(pCol, pBlock->info.rows, in, pResInfo->isNullRes);

  return pResInfo->numOfRes;
}

222 223 224
EFuncDataRequired countDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow) {
  SNode* pParam = nodesListGetNode(pFunc->pParameterList, 0);
  if (QUERY_NODE_COLUMN == nodeType(pParam) && PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)pParam)->colId) {
225
    return FUNC_DATA_REQUIRED_NOT_LOAD;
226
  }
227
  return FUNC_DATA_REQUIRED_STATIS_LOAD;
228
}
H
Haojun Liao 已提交
229 230 231 232 233 234

bool getCountFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(int64_t);
  return true;
}

5
54liuyao 已提交
235
static FORCE_INLINE int32_t getNumofElem(SqlFunctionCtx* pCtx) {
H
Haojun Liao 已提交
236 237 238
  int32_t numOfElem = 0;

  /*
H
Haojun Liao 已提交
239 240 241
   * 1. column data missing (schema modified) causes pInputCol->hasNull == true. pInput->colDataAggIsSet == true;
   * 2. for general non-primary key columns, pInputCol->hasNull may be true or false, pInput->colDataAggIsSet == true;
   * 3. for primary key column, pInputCol->hasNull always be false, pInput->colDataAggIsSet == false;
H
Haojun Liao 已提交
242 243
   */
  SInputColumnInfoData* pInput = &pCtx->input;
244
  SColumnInfoData*      pInputCol = pInput->pData[0];
H
Haojun Liao 已提交
245 246 247 248 249 250 251 252 253 254 255 256
  if (pInput->colDataAggIsSet && pInput->totalRows == pInput->numOfRows) {
    numOfElem = pInput->numOfRows - pInput->pColumnDataAgg[0]->numOfNull;
    ASSERT(numOfElem >= 0);
  } else {
    if (pInputCol->hasNull) {
      for (int32_t i = pInput->startRowIndex; i < pInput->startRowIndex + pInput->numOfRows; ++i) {
        if (colDataIsNull(pInputCol, pInput->totalRows, i, NULL)) {
          continue;
        }
        numOfElem += 1;
      }
    } else {
257 258
      // when counting on the primary time stamp column and no statistics data is presented, use the size value
      // directly.
H
Haojun Liao 已提交
259 260 261
      numOfElem = pInput->numOfRows;
    }
  }
5
54liuyao 已提交
262 263 264 265 266 267 268 269
  return numOfElem;
}
/*
 * count function does need the finalize, if data is missing, the default value, which is 0, is used
 * count function does not use the pCtx->interResBuf to keep the intermediate buffer
 */
int32_t countFunction(SqlFunctionCtx* pCtx) {
  int32_t numOfElem = getNumofElem(pCtx);
H
Haojun Liao 已提交
270
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
271 272
  char*                buf = GET_ROWCELL_INTERBUF(pResInfo);
  *((int64_t*)buf) += numOfElem;
H
Haojun Liao 已提交
273 274

  SET_VAL(pResInfo, numOfElem, 1);
wmmhello's avatar
wmmhello 已提交
275
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
276 277
}

5
54liuyao 已提交
278 279 280 281 282 283 284 285 286 287 288
int32_t countInvertFunction(SqlFunctionCtx* pCtx) {
  int32_t numOfElem = getNumofElem(pCtx);

  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
  char*                buf = GET_ROWCELL_INTERBUF(pResInfo);
  *((int64_t*)buf) -= numOfElem;

  SET_VAL(pResInfo, *((int64_t*)buf), 1);
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
289 290
#define LIST_ADD_N(_res, _col, _start, _rows, _t, numOfElem)             \
  do {                                                                   \
291
    _t* d = (_t*)(_col->pData);                                          \
H
Haojun Liao 已提交
292 293 294 295 296 297 298 299 300
    for (int32_t i = (_start); i < (_rows) + (_start); ++i) {            \
      if (((_col)->hasNull) && colDataIsNull_f((_col)->nullbitmap, i)) { \
        continue;                                                        \
      };                                                                 \
      (_res) += (d)[i];                                                  \
      (numOfElem)++;                                                     \
    }                                                                    \
  } while (0)

5
54liuyao 已提交
301 302 303 304 305 306 307 308 309 310 311 312
#define LIST_SUB_N(_res, _col, _start, _rows, _t, numOfElem)             \
  do {                                                                   \
    _t* d = (_t*)(_col->pData);                                          \
    for (int32_t i = (_start); i < (_rows) + (_start); ++i) {            \
      if (((_col)->hasNull) && colDataIsNull_f((_col)->nullbitmap, i)) { \
        continue;                                                        \
      };                                                                 \
      (_res) -= (d)[i];                                                  \
      (numOfElem)++;                                                     \
    }                                                                    \
  } while (0)

313
int32_t sumFunction(SqlFunctionCtx* pCtx) {
H
Haojun Liao 已提交
314 315 316 317
  int32_t numOfElem = 0;

  // Only the pre-computing information loaded and actual data does not loaded
  SInputColumnInfoData* pInput = &pCtx->input;
318 319
  SColumnDataAgg*       pAgg = pInput->pColumnDataAgg[0];
  int32_t               type = pInput->pData[0]->info.type;
H
Haojun Liao 已提交
320

321
  SSumRes* pSumRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
G
Ganlin Zhao 已提交
322

H
Haojun Liao 已提交
323 324 325 326 327
  if (pInput->colDataAggIsSet) {
    numOfElem = pInput->numOfRows - pAgg->numOfNull;
    ASSERT(numOfElem >= 0);

    if (IS_SIGNED_NUMERIC_TYPE(type)) {
328
      pSumRes->isum += pAgg->sum;
H
Haojun Liao 已提交
329
    } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
330
      pSumRes->usum += pAgg->sum;
H
Haojun Liao 已提交
331
    } else if (IS_FLOAT_TYPE(type)) {
332
      pSumRes->dsum += GET_DOUBLE_VAL((const char*)&(pAgg->sum));
H
Haojun Liao 已提交
333 334 335 336
    }
  } else {  // computing based on the true data block
    SColumnInfoData* pCol = pInput->pData[0];

337
    int32_t start = pInput->startRowIndex;
H
Haojun Liao 已提交
338 339
    int32_t numOfRows = pInput->numOfRows;

340 341
    if (IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_BOOL) {
      if (type == TSDB_DATA_TYPE_TINYINT || type == TSDB_DATA_TYPE_BOOL) {
342 343 344 345 346 347 348
        LIST_ADD_N(pSumRes->isum, pCol, start, numOfRows, int8_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_SMALLINT) {
        LIST_ADD_N(pSumRes->isum, pCol, start, numOfRows, int16_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_INT) {
        LIST_ADD_N(pSumRes->isum, pCol, start, numOfRows, int32_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_BIGINT) {
        LIST_ADD_N(pSumRes->isum, pCol, start, numOfRows, int64_t, numOfElem);
H
Haojun Liao 已提交
349
      }
350 351 352 353 354 355 356 357 358
    } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
      if (type == TSDB_DATA_TYPE_UTINYINT) {
        LIST_ADD_N(pSumRes->usum, pCol, start, numOfRows, uint8_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_USMALLINT) {
        LIST_ADD_N(pSumRes->usum, pCol, start, numOfRows, uint16_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_UINT) {
        LIST_ADD_N(pSumRes->usum, pCol, start, numOfRows, uint32_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_UBIGINT) {
        LIST_ADD_N(pSumRes->usum, pCol, start, numOfRows, uint64_t, numOfElem);
H
Haojun Liao 已提交
359
      }
360 361 362 363
    } else if (type == TSDB_DATA_TYPE_DOUBLE) {
      LIST_ADD_N(pSumRes->dsum, pCol, start, numOfRows, double, numOfElem);
    } else if (type == TSDB_DATA_TYPE_FLOAT) {
      LIST_ADD_N(pSumRes->dsum, pCol, start, numOfRows, float, numOfElem);
H
Haojun Liao 已提交
364 365 366 367 368
    }
  }

  // data in the check operation are all null, not output
  SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1);
wmmhello's avatar
wmmhello 已提交
369
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
370 371
}

5
54liuyao 已提交
372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430
int32_t sumInvertFunction(SqlFunctionCtx* pCtx) {
  int32_t numOfElem = 0;

  // Only the pre-computing information loaded and actual data does not loaded
  SInputColumnInfoData* pInput = &pCtx->input;
  SColumnDataAgg*       pAgg = pInput->pColumnDataAgg[0];
  int32_t               type = pInput->pData[0]->info.type;

  SSumRes* pSumRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));

  if (pInput->colDataAggIsSet) {
    numOfElem = pInput->numOfRows - pAgg->numOfNull;
    ASSERT(numOfElem >= 0);

    if (IS_SIGNED_NUMERIC_TYPE(type)) {
      pSumRes->isum -= pAgg->sum;
    } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
      pSumRes->usum -= pAgg->sum;
    } else if (IS_FLOAT_TYPE(type)) {
      pSumRes->dsum -= GET_DOUBLE_VAL((const char*)&(pAgg->sum));
    }
  } else {  // computing based on the true data block
    SColumnInfoData* pCol = pInput->pData[0];

    int32_t start = pInput->startRowIndex;
    int32_t numOfRows = pInput->numOfRows;

    if (IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_BOOL) {
      if (type == TSDB_DATA_TYPE_TINYINT || type == TSDB_DATA_TYPE_BOOL) {
        LIST_SUB_N(pSumRes->isum, pCol, start, numOfRows, int8_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_SMALLINT) {
        LIST_SUB_N(pSumRes->isum, pCol, start, numOfRows, int16_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_INT) {
        LIST_SUB_N(pSumRes->isum, pCol, start, numOfRows, int32_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_BIGINT) {
        LIST_SUB_N(pSumRes->isum, pCol, start, numOfRows, int64_t, numOfElem);
      }
    } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
      if (type == TSDB_DATA_TYPE_UTINYINT) {
        LIST_SUB_N(pSumRes->usum, pCol, start, numOfRows, uint8_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_USMALLINT) {
        LIST_SUB_N(pSumRes->usum, pCol, start, numOfRows, uint16_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_UINT) {
        LIST_SUB_N(pSumRes->usum, pCol, start, numOfRows, uint32_t, numOfElem);
      } else if (type == TSDB_DATA_TYPE_UBIGINT) {
        LIST_SUB_N(pSumRes->usum, pCol, start, numOfRows, uint64_t, numOfElem);
      }
    } else if (type == TSDB_DATA_TYPE_DOUBLE) {
      LIST_SUB_N(pSumRes->dsum, pCol, start, numOfRows, double, numOfElem);
    } else if (type == TSDB_DATA_TYPE_FLOAT) {
      LIST_SUB_N(pSumRes->dsum, pCol, start, numOfRows, float, numOfElem);
    }
  }

  // data in the check operation are all null, not output
  SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1);
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
431
bool getSumFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
H
Haojun Liao 已提交
432 433 434 435
  pEnv->calcMemSize = sizeof(SSumRes);
  return true;
}

G
Ganlin Zhao 已提交
436 437 438 439 440
bool getAvgFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(double);
  return true;
}

441
bool avgFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {
G
Ganlin Zhao 已提交
442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467
  if (!functionSetup(pCtx, pResultInfo)) {
    return false;
  }

  SAvgRes* pRes = GET_ROWCELL_INTERBUF(pResultInfo);
  memset(pRes, 0, sizeof(SAvgRes));
  return true;
}

int32_t avgFunction(SqlFunctionCtx* pCtx) {
  int32_t numOfElem = 0;

  // Only the pre-computing information loaded and actual data does not loaded
  SInputColumnInfoData* pInput = &pCtx->input;
  int32_t               type = pInput->pData[0]->info.type;

  SAvgRes* pAvgRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));

  // computing based on the true data block
  SColumnInfoData* pCol = pInput->pData[0];

  int32_t start = pInput->startRowIndex;
  int32_t numOfRows = pInput->numOfRows;

  switch (type) {
    case TSDB_DATA_TYPE_TINYINT: {
468 469 470 471
      int8_t* plist = (int8_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
G
Ganlin Zhao 已提交
472 473
        }

474 475 476
        numOfElem += 1;
        pAvgRes->count += 1;
        pAvgRes->sum.isum += plist[i];
G
Ganlin Zhao 已提交
477 478
      }

479 480 481 482
      break;
    }

    case TSDB_DATA_TYPE_SMALLINT: {
G
Ganlin Zhao 已提交
483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561
      int16_t* plist = (int16_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pAvgRes->count += 1;
        pAvgRes->sum.isum += plist[i];
      }
      break;
    }

    case TSDB_DATA_TYPE_INT: {
      int32_t* plist = (int32_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pAvgRes->count += 1;
        pAvgRes->sum.isum += plist[i];
      }

      break;
    }

    case TSDB_DATA_TYPE_BIGINT: {
      int64_t* plist = (int64_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pAvgRes->count += 1;
        pAvgRes->sum.isum += plist[i];
      }
      break;
    }

    case TSDB_DATA_TYPE_FLOAT: {
      float* plist = (float*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pAvgRes->count += 1;
        pAvgRes->sum.dsum += plist[i];
      }
      break;
    }

    case TSDB_DATA_TYPE_DOUBLE: {
      double* plist = (double*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pAvgRes->count += 1;
        pAvgRes->sum.dsum += plist[i];
      }
      break;
    }

    default:
      break;
  }

  // data in the check operation are all null, not output
  SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1);
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624
#define LIST_AVG_N(sumT, T)                                                   \
  do {                                                                        \
      T* plist = (T*)pCol->pData;                                             \
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {   \
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {          \
          continue;                                                           \
        }                                                                     \
                                                                              \
        numOfElem += 1;                                                       \
        pAvgRes->count -= 1;                                                  \
        sumT -= plist[i];                                                     \
      }                                                                       \
  } while (0)

int32_t avgInvertFunction(SqlFunctionCtx* pCtx) {
  int32_t numOfElem = 0;

  // Only the pre-computing information loaded and actual data does not loaded
  SInputColumnInfoData* pInput = &pCtx->input;
  int32_t               type = pInput->pData[0]->info.type;

  SAvgRes* pAvgRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));

  // computing based on the true data block
  SColumnInfoData* pCol = pInput->pData[0];

  int32_t start = pInput->startRowIndex;
  int32_t numOfRows = pInput->numOfRows;

  switch (type) {
    case TSDB_DATA_TYPE_TINYINT: {
      LIST_AVG_N(pAvgRes->sum.isum, int8_t);
      break;
    }
    case TSDB_DATA_TYPE_SMALLINT: {
      LIST_AVG_N(pAvgRes->sum.isum, int16_t);
      break;
    }
    case TSDB_DATA_TYPE_INT: {
      LIST_AVG_N(pAvgRes->sum.isum, int32_t);
      break;
    }
    case TSDB_DATA_TYPE_BIGINT: {
      LIST_AVG_N(pAvgRes->sum.isum, int64_t);
      break;
    }
    case TSDB_DATA_TYPE_FLOAT: {
      LIST_AVG_N(pAvgRes->sum.dsum, float);
      break;
    }
    case TSDB_DATA_TYPE_DOUBLE: {
      LIST_AVG_N(pAvgRes->sum.dsum, double);
      break;
    }
    default:
      break;
  }

  // data in the check operation are all null, not output
  SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1);
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
625
int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
G
Ganlin Zhao 已提交
626
  SInputColumnInfoData* pInput = &pCtx->input;
627 628
  int32_t               type = pInput->pData[0]->info.type;
  SAvgRes*              pAvgRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
G
Ganlin Zhao 已提交
629
  if (IS_INTEGER_TYPE(type)) {
630
    pAvgRes->result = pAvgRes->sum.isum / ((double)pAvgRes->count);
G
Ganlin Zhao 已提交
631
  } else {
632
    pAvgRes->result = pAvgRes->sum.dsum / ((double)pAvgRes->count);
G
Ganlin Zhao 已提交
633
  }
634

H
Haojun Liao 已提交
635
  return functionFinalize(pCtx, pBlock);
G
Ganlin Zhao 已提交
636 637
}

638
EFuncDataRequired statisDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow) {
639 640 641
  return FUNC_DATA_REQUIRED_STATIS_LOAD;
}

642 643 644 645 646
typedef struct SMinmaxResInfo {
  bool      assign;   // assign the first value or not
  int64_t   v;
  STuplePos tuplePos;
} SMinmaxResInfo;
647

648
bool minmaxFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {
649 650 651 652
  if (!functionSetup(pCtx, pResultInfo)) {
    return false;  // not initialized since it has been initialized
  }

653 654 655
  SMinmaxResInfo* buf = GET_ROWCELL_INTERBUF(pResultInfo);
  buf->assign = false;
  buf->tuplePos.pageId = -1;
656 657 658
  return true;
}

H
Haojun Liao 已提交
659
bool getMinmaxFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
660
  pEnv->calcMemSize = sizeof(SMinmaxResInfo);
661 662 663 664 665 666 667 668 669
  return true;
}

#define GET_TS_LIST(x)    ((TSKEY*)((x)->ptsList))
#define GET_TS_DATA(x, y) (GET_TS_LIST(x)[(y)])

#define DO_UPDATE_TAG_COLUMNS_WITHOUT_TS(ctx)                      \
  do {                                                             \
    for (int32_t _i = 0; _i < (ctx)->tagInfo.numOfTagCols; ++_i) { \
670
      SqlFunctionCtx* __ctx = (ctx)->tagInfo.pTagCtxList[_i];      \
H
Haojun Liao 已提交
671
      __ctx->fpSet.process(__ctx);                                 \
672 673 674
    }                                                              \
  } while (0);

675 676
#define DO_UPDATE_SUBSID_RES(ctx, ts)                          \
  do {                                                         \
677
    for (int32_t _i = 0; _i < (ctx)->subsidiaries.num; ++_i) { \
678 679 680 681 682 683 684
      SqlFunctionCtx* __ctx = (ctx)->subsidiaries.pCtx[_i];    \
      if (__ctx->functionId == FUNCTION_TS_DUMMY) {            \
        __ctx->tag.i = (ts);                                   \
        __ctx->tag.nType = TSDB_DATA_TYPE_BIGINT;              \
      }                                                        \
      __ctx->fpSet.process(__ctx);                             \
    }                                                          \
685 686 687
  } while (0)

#define UPDATE_DATA(ctx, left, right, num, sign, _ts) \
H
Haojun Liao 已提交
688 689 690 691 692 693
  do {                                                \
    if (((left) < (right)) ^ (sign)) {                \
      (left) = (right);                               \
      DO_UPDATE_SUBSID_RES(ctx, _ts);                 \
      (num) += 1;                                     \
    }                                                 \
694 695
  } while (0)

H
Haojun Liao 已提交
696
#define LOOPCHECK_N(val, _col, ctx, _t, _nrow, _start, sign, num)        \
697
  do {                                                                   \
698
    _t* d = (_t*)((_col)->pData);                                        \
699 700 701 702 703 704 705 706 707
    for (int32_t i = (_start); i < (_nrow) + (_start); ++i) {            \
      if (((_col)->hasNull) && colDataIsNull_f((_col)->nullbitmap, i)) { \
        continue;                                                        \
      }                                                                  \
      TSKEY ts = (ctx)->ptsList != NULL ? GET_TS_DATA(ctx, i) : 0;       \
      UPDATE_DATA(ctx, val, d[i], num, sign, ts);                        \
    }                                                                    \
  } while (0)

708 709 710
static void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);
static void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos);

711
int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
712 713 714
  int32_t numOfElems = 0;

  SInputColumnInfoData* pInput = &pCtx->input;
715
  SColumnDataAgg*       pAgg = pInput->pColumnDataAgg[0];
716 717

  SColumnInfoData* pCol = pInput->pData[0];
718
  int32_t          type = pCol->info.type;
719 720

  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
721
  SMinmaxResInfo *pBuf = GET_ROWCELL_INTERBUF(pResInfo);
722 723 724 725 726 727 728 729 730 731 732 733 734

  // data in current data block are qualified to the query
  if (pInput->colDataAggIsSet) {
    numOfElems = pInput->numOfRows - pAgg->numOfNull;
    ASSERT(pInput->numOfRows == pInput->totalRows && numOfElems >= 0);
    if (numOfElems == 0) {
      return numOfElems;
    }

    void*   tval = NULL;
    int16_t index = 0;

    if (isMinFunc) {
735
      tval = &pInput->pColumnDataAgg[0]->min;
736 737
      index = pInput->pColumnDataAgg[0]->minIndex;
    } else {
738
      tval = &pInput->pColumnDataAgg[0]->max;
739 740 741
      index = pInput->pColumnDataAgg[0]->maxIndex;
    }

742
    // the index is the original position, not the relative position
743
    TSKEY key = (pCtx->ptsList != NULL) ? pCtx->ptsList[index] : TSKEY_INITIAL_VAL;
744

745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769
    if (!pBuf->assign) {
      pBuf->v = *(int64_t*)tval;
      if (pCtx->subsidiaries.num > 0) {
        saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
      }
    } else {
      if (IS_SIGNED_NUMERIC_TYPE(type)) {
        int64_t prev = 0;
        GET_TYPED_DATA(prev, int64_t, type, &pBuf->v);

        int64_t val = GET_INT64_VAL(tval);
        if ((prev < val) ^ isMinFunc) {
          pBuf->v = val;
          //        for (int32_t i = 0; i < (pCtx)->subsidiaries.num; ++i) {
          //          SqlFunctionCtx* __ctx = pCtx->subsidiaries.pCtx[i];
          //          if (__ctx->functionId == FUNCTION_TS_DUMMY) {  // TODO refactor
          //            __ctx->tag.i = key;
          //            __ctx->tag.nType = TSDB_DATA_TYPE_BIGINT;
          //          }
          //
          //          __ctx->fpSet.process(__ctx);
          //        }

          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
770
          }
771
        }
772

773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791
      } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
        uint64_t prev = 0;
        GET_TYPED_DATA(prev, uint64_t, type, &pBuf->v);

        uint64_t val = GET_UINT64_VAL(tval);
        if ((prev < val) ^ isMinFunc) {
          pBuf->v = val;
          //          for (int32_t i = 0; i < (pCtx)->subsidiaries.num; ++i) {
          //            SqlFunctionCtx* __ctx = pCtx->subsidiaries.pCtx[i];
          //            if (__ctx->functionId == FUNCTION_TS_DUMMY) {  // TODO refactor
          //              __ctx->tag.i = key;
          //              __ctx->tag.nType = TSDB_DATA_TYPE_BIGINT;
          //            }
          //
          //            __ctx->fpSet.process(__ctx);
          //          }
          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
792
        }
793 794 795
      } else if (type == TSDB_DATA_TYPE_DOUBLE) {
        double prev = 0;
        GET_TYPED_DATA(prev, int64_t, type, &pBuf->v);
796

797 798 799 800 801 802
        double val = GET_DOUBLE_VAL(tval);
        if ((prev < val) ^ isMinFunc) {
          pBuf->v = val;

          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
H
Haojun Liao 已提交
803
          }
804 805 806 807
        }
      } else if (type == TSDB_DATA_TYPE_FLOAT) {
        double prev = 0;
        GET_TYPED_DATA(prev, int64_t, type, &pBuf->v);
808

809 810 811 812
        double val = GET_DOUBLE_VAL(tval);
        if ((prev < val) ^ isMinFunc) {
          pBuf->v = val;
        }
813

814 815
        if (pCtx->subsidiaries.num > 0) {
          saveTupleData(pCtx, index, pCtx->pSrcBlock, &pBuf->tuplePos);
H
Haojun Liao 已提交
816 817
        }
      }
818 819
    }

820
    pBuf->assign = true;
821 822 823 824 825 826
    return numOfElems;
  }

  int32_t start = pInput->startRowIndex;
  int32_t numOfRows = pInput->numOfRows;

827 828
  if (IS_SIGNED_NUMERIC_TYPE(type) || type == TSDB_DATA_TYPE_BOOL) {
    if (type == TSDB_DATA_TYPE_TINYINT || type == TSDB_DATA_TYPE_BOOL) {
829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858
      int8_t* pData = (int8_t*)pCol->pData;
      int8_t* val = (int8_t*)&pBuf->v;

      for (int32_t i = start; i < start + numOfRows; ++i) {
        if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        if (!pBuf->assign) {
          *val = pData[i];
          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
          pBuf->assign = true;
        } else {
          // ignore the equivalent data value
          if ((*val) == pData[i]) {
            continue;
          }

          if ((*val < pData[i]) ^ isMinFunc) {
            *val = pData[i];
            if (pCtx->subsidiaries.num > 0) {
              copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
            }
          }
        }

        numOfElems += 1;
      }
859
    } else if (type == TSDB_DATA_TYPE_SMALLINT) {
860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889
      int16_t* pData = (int16_t*)pCol->pData;
      int16_t* val = (int16_t*)&pBuf->v;

      for (int32_t i = start; i < start + numOfRows; ++i) {
        if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        if (!pBuf->assign) {
          *val = pData[i];
          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
          pBuf->assign = true;
        } else {
          // ignore the equivalent data value
          if ((*val) == pData[i]) {
            continue;
          }

          if ((*val < pData[i]) ^ isMinFunc) {
            *val = pData[i];
            if (pCtx->subsidiaries.num > 0) {
              copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
            }
          }
        }

        numOfElems += 1;
      }
890
    } else if (type == TSDB_DATA_TYPE_INT) {
891
      int32_t* pData = (int32_t*)pCol->pData;
892
      int32_t* val = (int32_t*)&pBuf->v;
893

H
Haojun Liao 已提交
894
      for (int32_t i = start; i < start + numOfRows; ++i) {
895 896 897 898
        if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

899
        if (!pBuf->assign) {
900
          *val = pData[i];
901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916
          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
          pBuf->assign = true;
        } else {
          // ignore the equivalent data value
          if ((*val) == pData[i]) {
            continue;
          }

          if ((*val < pData[i]) ^ isMinFunc) {
            *val = pData[i];
            if (pCtx->subsidiaries.num > 0) {
              copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
            }
          }
917 918 919 920
        }

        numOfElems += 1;
      }
921
    } else if (type == TSDB_DATA_TYPE_BIGINT) {
922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951
      int64_t* pData = (int64_t*)pCol->pData;
      int64_t* val = (int64_t*)&pBuf->v;

      for (int32_t i = start; i < start + numOfRows; ++i) {
        if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        if (!pBuf->assign) {
          *val = pData[i];
          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
          pBuf->assign = true;
        } else {
          // ignore the equivalent data value
          if ((*val) == pData[i]) {
            continue;
          }

          if ((*val < pData[i]) ^ isMinFunc) {
            *val = pData[i];
            if (pCtx->subsidiaries.num > 0) {
              copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
            }
          }
        }

        numOfElems += 1;
      }
952
    }
953 954
  } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
    if (type == TSDB_DATA_TYPE_UTINYINT) {
955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984
      uint8_t* pData = (uint8_t*)pCol->pData;
      uint8_t* val = (uint8_t*)&pBuf->v;

      for (int32_t i = start; i < start + numOfRows; ++i) {
        if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        if (!pBuf->assign) {
          *val = pData[i];
          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
          pBuf->assign = true;
        } else {
          // ignore the equivalent data value
          if ((*val) == pData[i]) {
            continue;
          }

          if ((*val < pData[i]) ^ isMinFunc) {
            *val = pData[i];
            if (pCtx->subsidiaries.num > 0) {
              copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
            }
          }
        }

        numOfElems += 1;
      }
985
    } else if (type == TSDB_DATA_TYPE_USMALLINT) {
986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015
      uint16_t* pData = (uint16_t*)pCol->pData;
      uint16_t* val = (uint16_t*)&pBuf->v;

      for (int32_t i = start; i < start + numOfRows; ++i) {
        if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        if (!pBuf->assign) {
          *val = pData[i];
          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
          pBuf->assign = true;
        } else {
          // ignore the equivalent data value
          if ((*val) == pData[i]) {
            continue;
          }

          if ((*val < pData[i]) ^ isMinFunc) {
            *val = pData[i];
            if (pCtx->subsidiaries.num > 0) {
              copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
            }
          }
        }

        numOfElems += 1;
      }
1016
    } else if (type == TSDB_DATA_TYPE_UINT) {
1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046
      uint32_t* pData = (uint32_t*)pCol->pData;
      uint32_t* val = (uint32_t*)&pBuf->v;

      for (int32_t i = start; i < start + numOfRows; ++i) {
        if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        if (!pBuf->assign) {
          *val = pData[i];
          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
          pBuf->assign = true;
        } else {
          // ignore the equivalent data value
          if ((*val) == pData[i]) {
            continue;
          }

          if ((*val < pData[i]) ^ isMinFunc) {
            *val = pData[i];
            if (pCtx->subsidiaries.num > 0) {
              copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
            }
          }
        }

        numOfElems += 1;
      }
1047
    } else if (type == TSDB_DATA_TYPE_UBIGINT) {
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 1073 1074 1075 1076 1077
      uint64_t* pData = (uint64_t*)pCol->pData;
      uint64_t* val = (uint64_t*)&pBuf->v;

      for (int32_t i = start; i < start + numOfRows; ++i) {
        if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        if (!pBuf->assign) {
          *val = pData[i];
          if (pCtx->subsidiaries.num > 0) {
            saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
          pBuf->assign = true;
        } else {
          // ignore the equivalent data value
          if ((*val) == pData[i]) {
            continue;
          }

          if ((*val < pData[i]) ^ isMinFunc) {
            *val = pData[i];
            if (pCtx->subsidiaries.num > 0) {
              copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
            }
          }
        }

        numOfElems += 1;
      }
1078
    }
1079
  } else if (type == TSDB_DATA_TYPE_DOUBLE) {
1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109
    double* pData = (double*)pCol->pData;
    double* val = (double*)&pBuf->v;

    for (int32_t i = start; i < start + numOfRows; ++i) {
      if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
        continue;
      }

      if (!pBuf->assign) {
        *val = pData[i];
        if (pCtx->subsidiaries.num > 0) {
          saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
        }
        pBuf->assign = true;
      } else {
        // ignore the equivalent data value
        if ((*val) == pData[i]) {
          continue;
        }

        if ((*val < pData[i]) ^ isMinFunc) {
          *val = pData[i];
          if (pCtx->subsidiaries.num > 0) {
            copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
        }
      }

      numOfElems += 1;
    }
1110
  } else if (type == TSDB_DATA_TYPE_FLOAT) {
1111
    float* pData = (float*)pCol->pData;
1112
    double* val = (double*)&pBuf->v;
1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140

    for (int32_t i = start; i < start + numOfRows; ++i) {
      if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
        continue;
      }

      if (!pBuf->assign) {
        *val = pData[i];
        if (pCtx->subsidiaries.num > 0) {
          saveTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
        }
        pBuf->assign = true;
      } else {
        // ignore the equivalent data value
        if ((*val) == pData[i]) {
          continue;
        }

        if ((*val < pData[i]) ^ isMinFunc) {
          *val = pData[i];
          if (pCtx->subsidiaries.num > 0) {
            copyTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
          }
        }
      }

      numOfElems += 1;
    }
1141 1142 1143
  }

  return numOfElems;
H
Haojun Liao 已提交
1144
}
1145

1146
int32_t minFunction(SqlFunctionCtx* pCtx) {
1147 1148
  int32_t numOfElems = doMinMaxHelper(pCtx, 1);
  SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
wmmhello's avatar
wmmhello 已提交
1149
  return TSDB_CODE_SUCCESS;
1150 1151
}

1152
int32_t maxFunction(SqlFunctionCtx* pCtx) {
1153 1154
  int32_t numOfElems = doMinMaxHelper(pCtx, 0);
  SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
wmmhello's avatar
wmmhello 已提交
1155
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1156 1157
}

1158 1159
static void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos *pTuplePos, int32_t rowIndex);

1160 1161 1162 1163 1164
int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
  SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);

  SMinmaxResInfo* pRes = GET_ROWCELL_INTERBUF(pEntryInfo);

1165
  int32_t type = pCtx->input.pData[0]->info.type;
1166 1167 1168 1169 1170 1171
  int32_t slotId = pCtx->pExpr->base.resSchema.slotId;

  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);

  // todo assign the tag value
  int32_t currentRow = pBlock->info.rows;
1172

1173 1174 1175 1176 1177 1178
  if (pCol->info.type == TSDB_DATA_TYPE_FLOAT) {
    float v = *(double*) &pRes->v;
    colDataAppend(pCol, currentRow, (const char*)&v, false);
  } else {
    colDataAppend(pCol, currentRow, (const char*)&pRes->v, false);
  }
1179

1180
  setSelectivityValue(pCtx, pBlock, &pRes->tuplePos, currentRow);
1181 1182
  return pEntryInfo->numOfRes;
}
1183

1184 1185 1186 1187
void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos *pTuplePos, int32_t rowIndex) {
  int32_t pageId = pTuplePos->pageId;
  int32_t offset = pTuplePos->offset;
  if (pTuplePos->pageId != -1) {
1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208
    SFilePage* pPage = getBufPage(pCtx->pBuf, pageId);

    bool* nullList = (bool*)((char*)pPage + offset);
    char* pStart = (char*)(nullList + pCtx->pSrcBlock->info.numOfCols * sizeof(bool));

    // todo set the offset value to optimize the performance.
    for (int32_t j = 0; j < pCtx->subsidiaries.num; ++j) {
      SqlFunctionCtx* pc = pCtx->subsidiaries.pCtx[j];

      SFunctParam* pFuncParam = &pc->pExpr->base.pParam[0];
      int32_t      srcSlotId = pFuncParam->pCol->slotId;
      int32_t      dstSlotId = pc->pExpr->base.resSchema.slotId;

      int32_t ps = 0;
      for (int32_t k = 0; k < srcSlotId; ++k) {
        SColumnInfoData* pSrcCol = taosArrayGet(pCtx->pSrcBlock->pDataBlock, k);
        ps += pSrcCol->info.bytes;
      }

      SColumnInfoData* pDstCol = taosArrayGet(pBlock->pDataBlock, dstSlotId);
      if (nullList[srcSlotId]) {
1209
        colDataAppendNULL(pDstCol, rowIndex);
1210
      } else {
1211
        colDataAppend(pDstCol, rowIndex, (pStart + ps), false);
1212 1213 1214 1215 1216
      }
    }
  }
}

H
Haojun Liao 已提交
1217 1218 1219 1220 1221
bool getStddevFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(SStddevRes);
  return true;
}

1222
bool stddevFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {
H
Haojun Liao 已提交
1223 1224 1225 1226 1227 1228 1229 1230 1231
  if (!functionSetup(pCtx, pResultInfo)) {
    return false;
  }

  SStddevRes* pRes = GET_ROWCELL_INTERBUF(pResultInfo);
  memset(pRes, 0, sizeof(SStddevRes));
  return true;
}

H
Haojun Liao 已提交
1232
int32_t stddevFunction(SqlFunctionCtx* pCtx) {
H
Haojun Liao 已提交
1233 1234 1235 1236
  int32_t numOfElem = 0;

  // Only the pre-computing information loaded and actual data does not loaded
  SInputColumnInfoData* pInput = &pCtx->input;
H
Haojun Liao 已提交
1237
  int32_t               type = pInput->pData[0]->info.type;
H
Haojun Liao 已提交
1238 1239 1240

  SStddevRes* pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));

H
Haojun Liao 已提交
1241 1242
  // computing based on the true data block
  SColumnInfoData* pCol = pInput->pData[0];
H
Haojun Liao 已提交
1243

H
Haojun Liao 已提交
1244 1245
  int32_t start = pInput->startRowIndex;
  int32_t numOfRows = pInput->numOfRows;
H
Haojun Liao 已提交
1246

H
Haojun Liao 已提交
1247 1248
  switch (type) {
    case TSDB_DATA_TYPE_TINYINT: {
1249 1250 1251 1252
      int8_t* plist = (int8_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + start; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
H
Haojun Liao 已提交
1253
        }
H
Haojun Liao 已提交
1254

1255 1256 1257 1258
        numOfElem += 1;
        pStddevRes->count += 1;
        pStddevRes->isum += plist[i];
        pStddevRes->quadraticISum += plist[i] * plist[i];
H
Haojun Liao 已提交
1259 1260
      }

1261 1262 1263 1264
      break;
    }

    case TSDB_DATA_TYPE_SMALLINT: {
H
Haojun Liao 已提交
1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318
      int16_t* plist = (int16_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pStddevRes->count += 1;
        pStddevRes->isum += plist[i];
        pStddevRes->quadraticISum += plist[i] * plist[i];
      }
      break;
    }

    case TSDB_DATA_TYPE_INT: {
      int32_t* plist = (int32_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pStddevRes->count += 1;
        pStddevRes->isum += plist[i];
        pStddevRes->quadraticISum += plist[i] * plist[i];
      }

      break;
    }

    case TSDB_DATA_TYPE_BIGINT: {
      int64_t* plist = (int64_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pStddevRes->count += 1;
        pStddevRes->isum += plist[i];
        pStddevRes->quadraticISum += plist[i] * plist[i];
      }
      break;
    }

    case TSDB_DATA_TYPE_FLOAT: {
      float* plist = (float*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pStddevRes->count += 1;
G
Ganlin Zhao 已提交
1319 1320
        pStddevRes->dsum += plist[i];
        pStddevRes->quadraticDSum += plist[i] * plist[i];
H
Haojun Liao 已提交
1321 1322 1323 1324
      }
      break;
    }

H
Haojun Liao 已提交
1325 1326 1327 1328 1329 1330 1331 1332 1333
    case TSDB_DATA_TYPE_DOUBLE: {
      double* plist = (double*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem += 1;
        pStddevRes->count += 1;
G
Ganlin Zhao 已提交
1334 1335
        pStddevRes->dsum += plist[i];
        pStddevRes->quadraticDSum += plist[i] * plist[i];
H
Haojun Liao 已提交
1336 1337 1338 1339 1340 1341 1342 1343
      }
      break;
    }

    default:
      break;
  }

H
Haojun Liao 已提交
1344 1345
  // data in the check operation are all null, not output
  SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1);
wmmhello's avatar
wmmhello 已提交
1346
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1347 1348
}

5
54liuyao 已提交
1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411
#define LIST_STDDEV_SUB_N(sumT, T)                                 \
  do {                                                             \
    T* plist = (T*)pCol->pData;                                    \
    for (int32_t i = start; i < numOfRows + start; ++i) {          \
      if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) { \
        continue;                                                  \
      }                                                            \
      numOfElem += 1;                                              \
      pStddevRes->count -= 1;                                      \
      sumT -= plist[i];                                            \
      pStddevRes->quadraticISum -= plist[i] * plist[i];            \
    }                                                              \
  } while (0)
  
int32_t stddevInvertFunction(SqlFunctionCtx* pCtx) {
  int32_t numOfElem = 0;

  // Only the pre-computing information loaded and actual data does not loaded
  SInputColumnInfoData* pInput = &pCtx->input;
  int32_t               type = pInput->pData[0]->info.type;

  SStddevRes* pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));

  // computing based on the true data block
  SColumnInfoData* pCol = pInput->pData[0];

  int32_t start = pInput->startRowIndex;
  int32_t numOfRows = pInput->numOfRows;

  switch (type) {
    case TSDB_DATA_TYPE_TINYINT: {
      LIST_STDDEV_SUB_N(pStddevRes->isum, int8_t);
      break;
    }
    case TSDB_DATA_TYPE_SMALLINT: {
      LIST_STDDEV_SUB_N(pStddevRes->isum, int16_t);
      break;
    }
    case TSDB_DATA_TYPE_INT: {
      LIST_STDDEV_SUB_N(pStddevRes->isum, int32_t);
      break;
    }
    case TSDB_DATA_TYPE_BIGINT: {
      LIST_STDDEV_SUB_N(pStddevRes->isum, int64_t);
      break;
    }
    case TSDB_DATA_TYPE_FLOAT: {
      LIST_STDDEV_SUB_N(pStddevRes->dsum, float);
      break;
    }
    case TSDB_DATA_TYPE_DOUBLE: {
      LIST_STDDEV_SUB_N(pStddevRes->dsum, double);
      break;
    }
    default:
      break;
  }

  // data in the check operation are all null, not output
  SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1);
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
1412
int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
G
Ganlin Zhao 已提交
1413
  SInputColumnInfoData* pInput = &pCtx->input;
1414 1415 1416
  int32_t               type = pInput->pData[0]->info.type;
  SStddevRes*           pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
  double                avg;
G
Ganlin Zhao 已提交
1417
  if (IS_INTEGER_TYPE(type)) {
1418 1419
    avg = pStddevRes->isum / ((double)pStddevRes->count);
    pStddevRes->result = sqrt(pStddevRes->quadraticISum / ((double)pStddevRes->count) - avg * avg);
G
Ganlin Zhao 已提交
1420
  } else {
1421 1422
    avg = pStddevRes->dsum / ((double)pStddevRes->count);
    pStddevRes->result = sqrt(pStddevRes->quadraticDSum / ((double)pStddevRes->count) - avg * avg);
G
Ganlin Zhao 已提交
1423
  }
1424

1425
  return functionFinalize(pCtx, pBlock);
H
Haojun Liao 已提交
1426 1427
}

1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439
bool getLeastSQRFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(SLeastSQRInfo);
  return true;
}

bool leastSQRFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {
  if (!functionSetup(pCtx, pResultInfo)) {
    return false;
  }

  SLeastSQRInfo* pInfo = GET_ROWCELL_INTERBUF(pResultInfo);

1440 1441 1442 1443
  pInfo->startVal = IS_FLOAT_TYPE(pCtx->param[1].param.nType) ? pCtx->param[1].param.d :
                                                                (double)pCtx->param[1].param.i;
  pInfo->stepVal = IS_FLOAT_TYPE(pCtx->param[1].param.nType) ? pCtx->param[2].param.d :
                                                                (double)pCtx->param[1].param.i;
1444 1445 1446 1447 1448 1449 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 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570
  return true;
}

#define LEASTSQR_CAL(p, x, y, index, step) \
  do {                                     \
    (p)[0][0] += (double)(x) * (x);        \
    (p)[0][1] += (double)(x);              \
    (p)[0][2] += (double)(x) * (y)[index]; \
    (p)[1][2] += (y)[index];               \
    (x) += step;                           \
  } while (0)

int32_t leastSQRFunction(SqlFunctionCtx* pCtx) {
  int32_t numOfElem = 0;

  SInputColumnInfoData* pInput = &pCtx->input;
  int32_t               type = pInput->pData[0]->info.type;

  SLeastSQRInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));

  SColumnInfoData* pCol = pInput->pData[0];

  double(*param)[3] = pInfo->matrix;
  double x = pInfo->startVal;

  int32_t start = pInput->startRowIndex;
  int32_t numOfRows = pInput->numOfRows;

  switch (type) {
    case TSDB_DATA_TYPE_TINYINT: {
      int8_t* plist = (int8_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }
        numOfElem++;
        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);

        break;
      }
    }
    case TSDB_DATA_TYPE_SMALLINT: {
      int16_t* plist = (int16_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem++;
        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
      }
      break;
    }

    case TSDB_DATA_TYPE_INT: {
      int32_t* plist = (int32_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem++;
        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
      }

      break;
    }

    case TSDB_DATA_TYPE_BIGINT: {
      int64_t* plist = (int64_t*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem++;
        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
      }
      break;
    }

    case TSDB_DATA_TYPE_FLOAT: {
      float* plist = (float*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem++;
        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
      }
      break;
    }

    case TSDB_DATA_TYPE_DOUBLE: {
      double* plist = (double*)pCol->pData;
      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
          continue;
        }

        numOfElem++;
        LEASTSQR_CAL(param, x, plist, i, pInfo->stepVal);
      }
      break;
    }

    default:
      break;
  }

  pInfo->startVal = x;
  pInfo->num += numOfElem;

  SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1);

  return TSDB_CODE_SUCCESS;
}

int32_t leastSQRFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
  SLeastSQRInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
  int32_t        slotId = pCtx->pExpr->base.resSchema.slotId;
  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);

  int32_t currentRow = pBlock->info.rows;

1571
  if (0 == pInfo->num) {
1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602
    return 0;
  }

  double(*param)[3] = pInfo->matrix;

  param[1][1] = (double)pInfo->num;
  param[1][0] = param[0][1];

  param[0][0] -= param[1][0] * (param[0][1] / param[1][1]);
  param[0][2] -= param[1][2] * (param[0][1] / param[1][1]);
  param[0][1] = 0;
  param[1][2] -= param[0][2] * (param[1][0] / param[0][0]);
  param[1][0] = 0;
  param[0][2] /= param[0][0];

  param[1][2] /= param[1][1];

  char buf[64] = {0};
  size_t len = snprintf(varDataVal(buf), sizeof(buf) - VARSTR_HEADER_SIZE, "{slop:%.6lf, intercept:%.6lf}", param[0][2], param[1][2]);
  varDataSetLen(buf, len);

  colDataAppend(pCol, currentRow, buf, false);

  return pResInfo->numOfRes;
}

int32_t leastSQRInvertFunction(SqlFunctionCtx* pCtx) {
  //TODO
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
1603 1604 1605 1606 1607
bool getPercentileFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(SPercentileInfo);
  return true;
}

1608
bool percentileFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo) {
H
Haojun Liao 已提交
1609 1610 1611 1612 1613
  if (!functionSetup(pCtx, pResultInfo)) {
    return false;
  }

  // in the first round, get the min-max value of all involved data
1614
  SPercentileInfo* pInfo = GET_ROWCELL_INTERBUF(pResultInfo);
H
Haojun Liao 已提交
1615 1616 1617 1618 1619
  SET_DOUBLE_VAL(&pInfo->minval, DBL_MAX);
  SET_DOUBLE_VAL(&pInfo->maxval, -DBL_MAX);
  pInfo->numOfElems = 0;

  return true;
H
Haojun Liao 已提交
1620 1621
}

1622 1623 1624
int32_t percentileFunction(SqlFunctionCtx* pCtx) {
  int32_t              notNullElems = 0;
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
1625 1626

  SInputColumnInfoData* pInput = &pCtx->input;
1627
  SColumnDataAgg*       pAgg = pInput->pColumnDataAgg[0];
H
Haojun Liao 已提交
1628

1629 1630
  SColumnInfoData* pCol = pInput->pData[0];
  int32_t          type = pCol->info.type;
1631

1632
  SPercentileInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo);
H
Haojun Liao 已提交
1633 1634
  if (pCtx->currentStage == REPEAT_SCAN && pInfo->stage == 0) {
    pInfo->stage += 1;
H
Haojun Liao 已提交
1635

H
Haojun Liao 已提交
1636 1637 1638
    // all data are null, set it completed
    if (pInfo->numOfElems == 0) {
      pResInfo->complete = true;
H
Haojun Liao 已提交
1639
      return 0;
H
Haojun Liao 已提交
1640 1641 1642 1643 1644 1645 1646
    } else {
      pInfo->pMemBucket = tMemBucketCreate(pCtx->inputBytes, pCtx->inputType, pInfo->minval, pInfo->maxval);
    }
  }

  // the first stage, only acquire the min/max value
  if (pInfo->stage == 0) {
1647
    if (pCtx->input.colDataAggIsSet) {
H
Haojun Liao 已提交
1648
      double tmin = 0.0, tmax = 0.0;
1649 1650 1651 1652 1653 1654 1655 1656 1657
      if (IS_SIGNED_NUMERIC_TYPE(type)) {
        tmin = (double)GET_INT64_VAL(&pAgg->min);
        tmax = (double)GET_INT64_VAL(&pAgg->max);
      } else if (IS_FLOAT_TYPE(type)) {
        tmin = GET_DOUBLE_VAL(&pAgg->min);
        tmax = GET_DOUBLE_VAL(&pAgg->max);
      } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
        tmin = (double)GET_UINT64_VAL(&pAgg->min);
        tmax = (double)GET_UINT64_VAL(&pAgg->max);
H
Haojun Liao 已提交
1658 1659 1660 1661 1662 1663 1664 1665 1666 1667
      }

      if (GET_DOUBLE_VAL(&pInfo->minval) > tmin) {
        SET_DOUBLE_VAL(&pInfo->minval, tmin);
      }

      if (GET_DOUBLE_VAL(&pInfo->maxval) < tmax) {
        SET_DOUBLE_VAL(&pInfo->maxval, tmax);
      }

1668
      pInfo->numOfElems += (pInput->numOfRows - pAgg->numOfNull);
H
Haojun Liao 已提交
1669
    } else {
1670 1671 1672 1673
      // check the valid data one by one
      int32_t start = pInput->startRowIndex;
      for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
        if (colDataIsNull_f(pCol->nullbitmap, i)) {
H
Haojun Liao 已提交
1674 1675 1676
          continue;
        }

1677
        char* data = colDataGetData(pCol, i);
1678

H
Haojun Liao 已提交
1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692
        double v = 0;
        GET_TYPED_DATA(v, double, pCtx->inputType, data);
        if (v < GET_DOUBLE_VAL(&pInfo->minval)) {
          SET_DOUBLE_VAL(&pInfo->minval, v);
        }

        if (v > GET_DOUBLE_VAL(&pInfo->maxval)) {
          SET_DOUBLE_VAL(&pInfo->maxval, v);
        }

        pInfo->numOfElems += 1;
      }
    }

H
Haojun Liao 已提交
1693
    return 0;
H
Haojun Liao 已提交
1694 1695 1696
  }

  // the second stage, calculate the true percentile value
1697 1698 1699
  int32_t start = pInput->startRowIndex;
  for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
    if (colDataIsNull_f(pCol->nullbitmap, i)) {
H
Haojun Liao 已提交
1700 1701 1702
      continue;
    }

1703
    char* data = colDataGetData(pCol, i);
1704

H
Haojun Liao 已提交
1705 1706 1707 1708
    notNullElems += 1;
    tMemBucketPut(pInfo->pMemBucket, data, 1);
  }

1709
  SET_VAL(pResInfo, notNullElems, 1);
wmmhello's avatar
wmmhello 已提交
1710
  return TSDB_CODE_SUCCESS;
1711 1712
}

1713
int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
1714
  SVariant* pVal = &pCtx->param[1].param;
1715
  double    v = pVal->nType == TSDB_DATA_TYPE_INT ? pVal->i : pVal->d;
H
Haojun Liao 已提交
1716

1717 1718
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
  SPercentileInfo*     ppInfo = (SPercentileInfo*)GET_ROWCELL_INTERBUF(pResInfo);
1719

1720
  tMemBucket* pMemBucket = ppInfo->pMemBucket;
1721 1722 1723 1724 1725
  if (pMemBucket != NULL && pMemBucket->total > 0) {  // check for null
    SET_DOUBLE_VAL(&ppInfo->result, getPercentile(pMemBucket, v));
  }

  tMemBucketDestroy(pMemBucket);
1726
  return functionFinalize(pCtx, pBlock);
H
Haojun Liao 已提交
1727
}
H
Haojun Liao 已提交
1728

H
Haojun Liao 已提交
1729 1730
bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
  SColumnNode* pNode = nodesListGetNode(pFunc->pParameterList, 0);
1731
  pEnv->calcMemSize = pNode->node.resType.bytes + sizeof(int64_t);
H
Haojun Liao 已提交
1732 1733 1734
  return true;
}

1735 1736 1737 1738 1739 1740 1741 1742
bool getSelectivityFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
  SColumnNode* pNode = nodesListGetNode(pFunc->pParameterList, 0);
  pEnv->calcMemSize = pNode->node.resType.bytes;
  return true;
}



1743 1744 1745 1746 1747
static FORCE_INLINE TSKEY getRowPTs(SColumnInfoData* pTsColInfo, int32_t rowIndex) {
  if (pTsColInfo == NULL) {
    return 0;
  }

1748
  return *(TSKEY*)colDataGetData(pTsColInfo, rowIndex);
1749 1750
}

1751 1752
// This ordinary first function does not care if current scan is ascending order or descending order scan
// the OPTIMIZED version of first function will only handle the ascending order scan
1753
int32_t firstFunction(SqlFunctionCtx* pCtx) {
H
Haojun Liao 已提交
1754 1755
  int32_t numOfElems = 0;

1756 1757
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
  char*                buf = GET_ROWCELL_INTERBUF(pResInfo);
H
Haojun Liao 已提交
1758 1759

  SInputColumnInfoData* pInput = &pCtx->input;
1760
  SColumnInfoData*      pInputCol = pInput->pData[0];
H
Haojun Liao 已提交
1761

1762 1763
  int32_t bytes = pInputCol->info.bytes;

H
Haojun Liao 已提交
1764
  // All null data column, return directly.
H
Haojun Liao 已提交
1765
  if (pInput->colDataAggIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
H
Haojun Liao 已提交
1766
    ASSERT(pInputCol->hasNull == true);
H
Haojun Liao 已提交
1767
    return 0;
H
Haojun Liao 已提交
1768 1769
  }

1770
  SColumnDataAgg* pColAgg = (pInput->colDataAggIsSet) ? pInput->pColumnDataAgg[0] : NULL;
1771

1772 1773
  TSKEY startKey = getRowPTs(pInput->pPTS, 0);
  TSKEY endKey = getRowPTs(pInput->pPTS, pInput->totalRows - 1);
1774

1775
  int32_t blockDataOrder = (startKey <= endKey) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790

  if (blockDataOrder == TSDB_ORDER_ASC) {
    // filter according to current result firstly
    if (pResInfo->numOfRes > 0) {
      TSKEY ts = *(TSKEY*)(buf + bytes);
      if (ts < startKey) {
        return TSDB_CODE_SUCCESS;
      }
    }

    for (int32_t i = pInput->startRowIndex; i < pInput->startRowIndex + pInput->numOfRows; ++i) {
      if (pInputCol->hasNull && colDataIsNull(pInputCol, pInput->totalRows, i, pColAgg)) {
        continue;
      }

1791 1792
      numOfElems++;

1793
      char* data = colDataGetData(pInputCol, i);
1794
      TSKEY cts = getRowPTs(pInput->pPTS, i);
1795

1796
      if (pResInfo->numOfRes == 0 || *(TSKEY*)(buf + bytes) > cts) {
1797 1798
        memcpy(buf, data, bytes);
        *(TSKEY*)(buf + bytes) = cts;
1799
        //        DO_UPDATE_TAG_COLUMNS(pCtx, ts);
1800 1801

        pResInfo->numOfRes = 1;
1802
        break;
1803 1804 1805 1806 1807 1808 1809 1810 1811 1812
      }
    }
  } else {
    // in case of descending order time stamp serial, which usually happens as the results of the nest query,
    // all data needs to be check.
    if (pResInfo->numOfRes > 0) {
      TSKEY ts = *(TSKEY*)(buf + bytes);
      if (ts < endKey) {
        return TSDB_CODE_SUCCESS;
      }
H
Haojun Liao 已提交
1813 1814
    }

1815 1816 1817 1818 1819
    for (int32_t i = pInput->numOfRows + pInput->startRowIndex - 1; i >= pInput->startRowIndex; --i) {
      if (pInputCol->hasNull && colDataIsNull(pInputCol, pInput->totalRows, i, pColAgg)) {
        continue;
      }

1820 1821
      numOfElems++;

1822
      char* data = colDataGetData(pInputCol, i);
1823
      TSKEY cts = getRowPTs(pInput->pPTS, i);
1824

1825
      if (pResInfo->numOfRes == 0 || *(TSKEY*)(buf + bytes) > cts) {
1826 1827
        memcpy(buf, data, bytes);
        *(TSKEY*)(buf + bytes) = cts;
1828
        //        DO_UPDATE_TAG_COLUMNS(pCtx, ts);
1829
        pResInfo->numOfRes = 1;
1830
        break;
1831 1832
      }
    }
H
Haojun Liao 已提交
1833 1834 1835
  }

  SET_VAL(pResInfo, numOfElems, 1);
wmmhello's avatar
wmmhello 已提交
1836
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1837 1838
}

1839
int32_t lastFunction(SqlFunctionCtx* pCtx) {
H
Haojun Liao 已提交
1840 1841
  int32_t numOfElems = 0;

1842 1843
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
  char*                buf = GET_ROWCELL_INTERBUF(pResInfo);
H
Haojun Liao 已提交
1844 1845

  SInputColumnInfoData* pInput = &pCtx->input;
1846
  SColumnInfoData*      pInputCol = pInput->pData[0];
H
Haojun Liao 已提交
1847

1848 1849
  int32_t bytes = pInputCol->info.bytes;

H
Haojun Liao 已提交
1850
  // All null data column, return directly.
1851
  if (pInput->colDataAggIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
H
Haojun Liao 已提交
1852
    ASSERT(pInputCol->hasNull == true);
H
Haojun Liao 已提交
1853
    return 0;
H
Haojun Liao 已提交
1854 1855
  }

1856
  SColumnDataAgg* pColAgg = (pInput->colDataAggIsSet) ? pInput->pColumnDataAgg[0] : NULL;
1857 1858 1859 1860

  TSKEY startKey = getRowPTs(pInput->pPTS, 0);
  TSKEY endKey = getRowPTs(pInput->pPTS, pInput->totalRows - 1);

1861
  int32_t blockDataOrder = (startKey <= endKey) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
1862 1863

  if (blockDataOrder == TSDB_ORDER_ASC) {
H
Haojun Liao 已提交
1864
    for (int32_t i = pInput->numOfRows + pInput->startRowIndex - 1; i >= pInput->startRowIndex; --i) {
1865
      if (pInputCol->hasNull && colDataIsNull(pInputCol, pInput->totalRows, i, pColAgg)) {
H
Haojun Liao 已提交
1866 1867 1868 1869
        continue;
      }

      numOfElems++;
1870 1871 1872

      char* data = colDataGetData(pInputCol, i);
      TSKEY cts = getRowPTs(pInput->pPTS, i);
1873
      if (pResInfo->numOfRes == 0 || *(TSKEY*)(buf + bytes) < cts) {
1874 1875 1876 1877 1878
        memcpy(buf, data, bytes);
        *(TSKEY*)(buf + bytes) = cts;
        //        DO_UPDATE_TAG_COLUMNS(pCtx, ts);
        pResInfo->numOfRes = 1;
      }
H
Haojun Liao 已提交
1879 1880
      break;
    }
1881
  } else {  // descending order
H
Haojun Liao 已提交
1882
    for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
1883
      if (pInputCol->hasNull && colDataIsNull(pInputCol, pInput->totalRows, i, pColAgg)) {
H
Haojun Liao 已提交
1884 1885 1886
        continue;
      }

1887
      numOfElems++;
H
Haojun Liao 已提交
1888

1889 1890
      char* data = colDataGetData(pInputCol, i);
      TSKEY cts = getRowPTs(pInput->pPTS, i);
1891
      if (pResInfo->numOfRes == 0 || *(TSKEY*)(buf + bytes) < cts) {
1892 1893 1894
        memcpy(buf, data, bytes);
        *(TSKEY*)(buf + bytes) = cts;
        pResInfo->numOfRes = 1;
1895
        //        DO_UPDATE_TAG_COLUMNS(pCtx, ts);
H
Haojun Liao 已提交
1896 1897 1898 1899 1900 1901
      }
      break;
    }
  }

  SET_VAL(pResInfo, numOfElems, 1);
wmmhello's avatar
wmmhello 已提交
1902
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1903
}
H
Haojun Liao 已提交
1904

H
Haojun Liao 已提交
1905 1906 1907 1908 1909
bool getDiffFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(SDiffInfo);
  return true;
}

1910
bool diffFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo) {
H
Haojun Liao 已提交
1911 1912 1913 1914 1915
  if (!functionSetup(pCtx, pResInfo)) {
    return false;
  }

  SDiffInfo* pDiffInfo = GET_ROWCELL_INTERBUF(pResInfo);
1916
  pDiffInfo->hasPrev = false;
H
Haojun Liao 已提交
1917
  pDiffInfo->prev.i64 = 0;
1918
  pDiffInfo->ignoreNegative = pCtx->param[1].param.i;  // TODO set correct param
H
Haojun Liao 已提交
1919 1920
  pDiffInfo->includeNull = false;
  pDiffInfo->firstOutput = false;
H
Haojun Liao 已提交
1921 1922 1923
  return true;
}

1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018
static void doSetPrevVal(SDiffInfo* pDiffInfo, int32_t type, const char* pv) {
  switch(type) {
    case TSDB_DATA_TYPE_BOOL:
    case TSDB_DATA_TYPE_TINYINT:
      pDiffInfo->prev.i64 = *(int8_t*) pv; break;
    case TSDB_DATA_TYPE_INT:
      pDiffInfo->prev.i64 = *(int32_t*) pv; break;
    case TSDB_DATA_TYPE_SMALLINT:
      pDiffInfo->prev.i64 = *(int16_t*) pv; break;
    case TSDB_DATA_TYPE_BIGINT:
      pDiffInfo->prev.i64 = *(int64_t*) pv; break;
    case TSDB_DATA_TYPE_FLOAT:
      pDiffInfo->prev.d64 = *(float *) pv; break;
    case TSDB_DATA_TYPE_DOUBLE:
      pDiffInfo->prev.d64 = *(double*) pv; break;
    default:
      ASSERT(0);
  }
}

static void doHandleDiff(SDiffInfo* pDiffInfo, int32_t type, const char* pv, SColumnInfoData* pOutput, int32_t pos, int32_t order) {
    int32_t factor = (order == TSDB_ORDER_ASC)? 1:-1;
  switch (type) {
    case TSDB_DATA_TYPE_INT: {
      int32_t v = *(int32_t*)pv;
      int32_t delta = factor*(v - pDiffInfo->prev.i64);  // direct previous may be null
      if (delta < 0 && pDiffInfo->ignoreNegative) {
        colDataSetNull_f(pOutput->nullbitmap, pos);
      } else {
        colDataAppendInt32(pOutput, pos, &delta);
      }
      pDiffInfo->prev.i64 = v;
      break;
    }
    case TSDB_DATA_TYPE_BOOL:
    case TSDB_DATA_TYPE_TINYINT: {
      int8_t v = *(int8_t*)pv;
      int8_t delta = factor*(v - pDiffInfo->prev.i64);  // direct previous may be null
      if (delta < 0 && pDiffInfo->ignoreNegative) {
        colDataSetNull_f(pOutput->nullbitmap, pos);
      } else {
        colDataAppendInt8(pOutput, pos, &delta);
      }
      pDiffInfo->prev.i64 = v;
      break;
    }
    case TSDB_DATA_TYPE_SMALLINT: {
      int16_t v = *(int16_t*)pv;
      int16_t delta = factor*(v - pDiffInfo->prev.i64);  // direct previous may be null
      if (delta < 0 && pDiffInfo->ignoreNegative) {
        colDataSetNull_f(pOutput->nullbitmap, pos);
      } else {
        colDataAppendInt16(pOutput, pos, &delta);
      }
      pDiffInfo->prev.i64 = v;
      break;
    }
    case TSDB_DATA_TYPE_BIGINT: {
      int64_t v = *(int64_t*)pv;
      int64_t delta = factor*(v - pDiffInfo->prev.i64);  // direct previous may be null
      if (delta < 0 && pDiffInfo->ignoreNegative) {
        colDataSetNull_f(pOutput->nullbitmap, pos);
      } else {
        colDataAppendInt64(pOutput, pos, &delta);
      }
      pDiffInfo->prev.i64 = v;
      break;
    }
    case TSDB_DATA_TYPE_FLOAT: {
      float v = *(float*)pv;
      float delta = factor*(v - pDiffInfo->prev.d64);  // direct previous may be null
      if (delta < 0 && pDiffInfo->ignoreNegative) {
        colDataSetNull_f(pOutput->nullbitmap, pos);
      } else {
        colDataAppendFloat(pOutput, pos, &delta);
      }
      pDiffInfo->prev.d64 = v;
      break;
    }
    case TSDB_DATA_TYPE_DOUBLE: {
      double v = *(double*)pv;
      double delta = factor*(v - pDiffInfo->prev.d64);  // direct previous may be null
      if (delta < 0 && pDiffInfo->ignoreNegative) {
        colDataSetNull_f(pOutput->nullbitmap, pos);
      } else {
        colDataAppendDouble(pOutput, pos, &delta);
      }
      pDiffInfo->prev.d64 = v;
      break;
    }
    default:
      ASSERT(0);
  }
  }

2019 2020 2021
int32_t diffFunction(SqlFunctionCtx* pCtx) {
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
  SDiffInfo*           pDiffInfo = GET_ROWCELL_INTERBUF(pResInfo);
H
Haojun Liao 已提交
2022 2023 2024

  SInputColumnInfoData* pInput = &pCtx->input;

2025
  SColumnInfoData* pInputCol = pInput->pData[0];
H
Haojun Liao 已提交
2026
  SColumnInfoData* pTsOutput = pCtx->pTsOutput;
H
Haojun Liao 已提交
2027

2028 2029
  int32_t numOfElems = 0;
  TSKEY* tsList = (int64_t*)pInput->pPTS->pData;
H
Haojun Liao 已提交
2030
  int32_t startOffset = pCtx->offset;
H
Haojun Liao 已提交
2031

2032
  SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;
H
Haojun Liao 已提交
2033

2034 2035 2036
  if (pCtx->order == TSDB_ORDER_ASC) {
    for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
      int32_t pos = startOffset + numOfElems;
H
Haojun Liao 已提交
2037

2038 2039 2040 2041 2042
      if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
        if (pDiffInfo->includeNull) {
          colDataSetNull_f(pOutput->nullbitmap, pos);
          if (tsList != NULL) {
            colDataAppendInt64(pTsOutput, pos, &tsList[i]);
H
Haojun Liao 已提交
2043 2044
          }

2045
          numOfElems += 1;
H
Haojun Liao 已提交
2046
        }
2047
        continue;
H
Haojun Liao 已提交
2048 2049
      }

2050
      char* pv = colDataGetData(pInputCol, i);
H
Haojun Liao 已提交
2051

2052 2053 2054 2055
      if (pDiffInfo->hasPrev) {
        doHandleDiff(pDiffInfo, pInputCol->info.type, pv, pOutput, pos, pCtx->order);
        if (pTsOutput != NULL) {
          colDataAppendInt64(pTsOutput, pos, &tsList[i]);
H
Haojun Liao 已提交
2056 2057 2058
        }

        numOfElems++;
2059 2060
      } else {
        doSetPrevVal(pDiffInfo, pInputCol->info.type, pv);
H
Haojun Liao 已提交
2061 2062
      }

2063 2064 2065 2066 2067 2068 2069 2070 2071 2072 2073 2074
      pDiffInfo->hasPrev = true;
    }
  } else {
    for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
      int32_t pos = startOffset + numOfElems;

      if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
        if (pDiffInfo->includeNull) {
          colDataSetNull_f(pOutput->nullbitmap, pos);
          if (tsList != NULL) {
            colDataAppendInt64(pTsOutput, pos, &tsList[i]);
          }
H
Haojun Liao 已提交
2075

2076
          numOfElems += 1;
H
Haojun Liao 已提交
2077
        }
2078
        continue;
H
Haojun Liao 已提交
2079 2080
      }

2081
      char* pv = colDataGetData(pInputCol, i);
H
Haojun Liao 已提交
2082

2083 2084 2085 2086 2087
      // there is a row of previous data block to be handled in the first place.
      if (pDiffInfo->hasPrev) {
        doHandleDiff(pDiffInfo, pInputCol->info.type, pv, pOutput, pos, pCtx->order);
        if (pTsOutput != NULL) {
          colDataAppendInt64(pTsOutput, pos, &pDiffInfo->prevTs);
H
Haojun Liao 已提交
2088 2089 2090
        }

        numOfElems++;
2091 2092
      } else {
        doSetPrevVal(pDiffInfo, pInputCol->info.type, pv);
H
Haojun Liao 已提交
2093 2094
      }

2095 2096 2097
      pDiffInfo->hasPrev = true;
      if (pTsOutput != NULL) {
        pDiffInfo->prevTs = tsList[i];
H
Haojun Liao 已提交
2098 2099 2100 2101 2102
      }
    }
  }

  // initial value is not set yet
2103
  return numOfElems;
H
Haojun Liao 已提交
2104
}
H
Haojun Liao 已提交
2105

2106
bool getTopBotFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
2107
  SValueNode* pkNode = (SValueNode*)nodesListGetNode(pFunc->pParameterList, 1);
2108
  pEnv->calcMemSize = sizeof(STopBotRes) + pkNode->datum.i * sizeof(STopBotResItem);
2109 2110 2111
  return true;
}

2112 2113 2114 2115
static STopBotRes* getTopBotOutputInfo(SqlFunctionCtx* pCtx) {
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
  STopBotRes*          pRes = GET_ROWCELL_INTERBUF(pResInfo);
  pRes->pItems = (STopBotResItem*)((char*)pRes + sizeof(STopBotRes));
2116 2117

  return pRes;
2118 2119
}

2120
static void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type,
2121
                            uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery);
2122

2123 2124 2125
int32_t topFunction(SqlFunctionCtx* pCtx) {
  int32_t              numOfElems = 0;
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
2126 2127

  SInputColumnInfoData* pInput = &pCtx->input;
2128
  SColumnInfoData*      pCol = pInput->pData[0];
2129 2130 2131 2132

  int32_t type = pInput->pData[0]->info.type;

  int32_t start = pInput->startRowIndex;
2133
  for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
2134 2135 2136
    if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
      continue;
    }
2137

2138
    numOfElems++;
2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154 2155 2156 2157 2158 2159
    char* data = colDataGetData(pCol, i);
    doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, type, pInput->uid, pResInfo, true);
  }

  return TSDB_CODE_SUCCESS;
}

int32_t bottomFunction(SqlFunctionCtx* pCtx) {
  int32_t              numOfElems = 0;
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);

  SInputColumnInfoData* pInput = &pCtx->input;
  SColumnInfoData*      pCol = pInput->pData[0];

  int32_t type = pInput->pData[0]->info.type;

  int32_t start = pInput->startRowIndex;
  for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
    if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
      continue;
    }
2160

2161
    numOfElems++;
2162
    char* data = colDataGetData(pCol, i);
2163
    doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, type, pInput->uid, pResInfo, false);
2164 2165
  }

2166
  return TSDB_CODE_SUCCESS;
2167 2168
}

2169 2170
static int32_t topBotResComparFn(const void* p1, const void* p2, const void* param) {
  uint16_t type = *(uint16_t*)param;
2171

2172 2173
  STopBotResItem* val1 = (STopBotResItem*)p1;
  STopBotResItem* val2 = (STopBotResItem*)p2;
2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190 2191 2192 2193 2194 2195

  if (IS_SIGNED_NUMERIC_TYPE(type)) {
    if (val1->v.i == val2->v.i) {
      return 0;
    }

    return (val1->v.i > val2->v.i) ? 1 : -1;
  } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
    if (val1->v.u == val2->v.u) {
      return 0;
    }

    return (val1->v.u > val2->v.u) ? 1 : -1;
  }

  if (val1->v.d == val2->v.d) {
    return 0;
  }

  return (val1->v.d > val2->v.d) ? 1 : -1;
}

2196
void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type,
2197
                     uint64_t uid, SResultRowEntryInfo* pEntryInfo, bool isTopQuery) {
2198 2199
  STopBotRes* pRes = getTopBotOutputInfo(pCtx);
  int32_t     maxSize = pCtx->param[1].param.i;
2200

2201 2202 2203
  SVariant val = {0};
  taosVariantCreateFromBinary(&val, pData, tDataTypes[type].bytes, type);

2204
  STopBotResItem* pItems = pRes->pItems;
2205 2206 2207
  assert(pItems != NULL);

  // not full yet
2208 2209
  if (pEntryInfo->numOfRes < maxSize) {
    STopBotResItem* pItem = &pItems[pEntryInfo->numOfRes];
2210
    pItem->v = val;
2211
    pItem->uid = uid;
2212

2213
    // save the data of this tuple
2214
    saveTupleData(pCtx, rowIndex, pSrcBlock, &pItem->tuplePos);
2215 2216 2217

    // allocate the buffer and keep the data of this row into the new allocated buffer
    pEntryInfo->numOfRes++;
2218
    taosheapsort((void*)pItems, sizeof(STopBotResItem), pEntryInfo->numOfRes, (const void*)&type, topBotResComparFn,
2219
                 !isTopQuery);
2220
  } else {  // replace the minimum value in the result
2221 2222 2223 2224 2225 2226 2227 2228 2229
    if ((isTopQuery && (
        (IS_SIGNED_NUMERIC_TYPE(type) && val.i > pItems[0].v.i) ||
        (IS_UNSIGNED_NUMERIC_TYPE(type) && val.u > pItems[0].v.u) ||
        (IS_FLOAT_TYPE(type) && val.d > pItems[0].v.d)))
        || (!isTopQuery && (
        (IS_SIGNED_NUMERIC_TYPE(type) && val.i < pItems[0].v.i) ||
        (IS_UNSIGNED_NUMERIC_TYPE(type) && val.u < pItems[0].v.u) ||
        (IS_FLOAT_TYPE(type) && val.d < pItems[0].v.d))
        )) {
2230
      // replace the old data and the coresponding tuple data
2231
      STopBotResItem* pItem = &pItems[0];
2232
      pItem->v = val;
2233
      pItem->uid = uid;
2234 2235

      // save the data of this tuple by over writing the old data
2236
      copyTupleData(pCtx, rowIndex, pSrcBlock, &pItem->tuplePos);
2237
      taosheapadjust((void*)pItems, sizeof(STopBotResItem), 0, pEntryInfo->numOfRes - 1, (const void*)&type,
2238
                     topBotResComparFn, NULL, !isTopQuery);
2239
    }
2240 2241
  }
}
2242

2243
void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos) {
2244 2245 2246 2247 2248 2249 2250 2251 2252 2253 2254 2255 2256 2257 2258
  SFilePage* pPage = NULL;

  int32_t completeRowSize = pSrcBlock->info.rowSize + pSrcBlock->info.numOfCols * sizeof(bool);

  if (pCtx->curBufPage == -1) {
    pPage = getNewBufPage(pCtx->pBuf, 0, &pCtx->curBufPage);
    pPage->num = sizeof(SFilePage);
  } else {
    pPage = getBufPage(pCtx->pBuf, pCtx->curBufPage);
    if (pPage->num + completeRowSize > getBufPageSize(pCtx->pBuf)) {
      pPage = getNewBufPage(pCtx->pBuf, 0, &pCtx->curBufPage);
      pPage->num = sizeof(SFilePage);
    }
  }

2259
  pPos->pageId = pCtx->curBufPage;
2260 2261 2262 2263 2264 2265 2266 2267 2268 2269

  // keep the current row data, extract method
  int32_t offset = 0;
  bool*   nullList = (bool*)((char*)pPage + pPage->num);
  char*   pStart = (char*)(nullList + sizeof(bool) * pSrcBlock->info.numOfCols);
  for (int32_t i = 0; i < pSrcBlock->info.numOfCols; ++i) {
    SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, i);
    bool             isNull = colDataIsNull_s(pCol, rowIndex);
    if (isNull) {
      nullList[i] = true;
2270
      offset += pCol->info.bytes;
2271 2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282 2283
      continue;
    }

    char* p = colDataGetData(pCol, rowIndex);
    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
      memcpy(pStart + offset, p, varDataTLen(p));
    } else {
      memcpy(pStart + offset, p, pCol->info.bytes);
    }

    offset += pCol->info.bytes;
  }

2284
  pPos->offset = pPage->num;
2285 2286 2287 2288 2289 2290
  pPage->num += completeRowSize;

  setBufPageDirty(pPage, true);
  releaseBufPage(pCtx->pBuf, pPage);
}

2291 2292
void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos) {
  SFilePage* pPage = getBufPage(pCtx->pBuf, pPos->pageId);
2293

2294
  bool* nullList = (bool*)((char*)pPage + pPos->offset);
2295 2296 2297
  char* pStart = (char*)(nullList + pSrcBlock->info.numOfCols * sizeof(bool));

  int32_t offset = 0;
2298
  for (int32_t i = 0; i < pSrcBlock->info.numOfCols; ++i) {
2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318
    SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, i);
    if ((nullList[i] = colDataIsNull_s(pCol, rowIndex)) == true) {
      continue;
    }

    char* p = colDataGetData(pCol, rowIndex);
    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
      memcpy(pStart + offset, p, varDataTLen(p));
    } else {
      memcpy(pStart + offset, p, pCol->info.bytes);
    }

    offset += pCol->info.bytes;
  }

  setBufPageDirty(pPage, true);
  releaseBufPage(pCtx->pBuf, pPage);
}

int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
2319 2320
  SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(pCtx);
  STopBotRes*          pRes = GET_ROWCELL_INTERBUF(pEntryInfo);
2321 2322
  pEntryInfo->complete = true;

2323 2324 2325
  int32_t type = pCtx->input.pData[0]->info.type;
  int32_t slotId = pCtx->pExpr->base.resSchema.slotId;

2326 2327 2328 2329
  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);

  // todo assign the tag value and the corresponding row data
  int32_t currentRow = pBlock->info.rows;
2330 2331 2332 2333 2334 2335 2336
  for (int32_t i = 0; i < pEntryInfo->numOfRes; ++i) {
    STopBotResItem* pItem = &pRes->pItems[i];
    if (type == TSDB_DATA_TYPE_FLOAT) {
      float v = pItem->v.d;
      colDataAppend(pCol, currentRow, (const char*)&v, false);
    } else {
      colDataAppend(pCol, currentRow, (const char*)&pItem->v.i, false);
2337
    }
2338 2339 2340

    setSelectivityValue(pCtx, pBlock, &pRes->pItems[i].tuplePos, currentRow);
    currentRow += 1;
2341 2342 2343
  }

  return pEntryInfo->numOfRes;
2344
}
G
Ganlin Zhao 已提交
2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425 2426 2427 2428 2429 2430 2431 2432 2433 2434 2435 2436 2437 2438 2439 2440 2441 2442

bool getSpreadFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(SSpreadInfo);
  return true;
}

bool spreadFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo) {
  if (!functionSetup(pCtx, pResultInfo)) {
    return false;
  }

  SSpreadInfo* pInfo = GET_ROWCELL_INTERBUF(pResultInfo);
  SET_DOUBLE_VAL(&pInfo->min, DBL_MAX);
  SET_DOUBLE_VAL(&pInfo->max, -DBL_MAX);
  pInfo->hasResult = false;
  return true;
}

int32_t spreadFunction(SqlFunctionCtx *pCtx) {
  int32_t numOfElems = 0;

  // Only the pre-computing information loaded and actual data does not loaded
  SInputColumnInfoData* pInput = &pCtx->input;
  SColumnDataAgg *pAgg = pInput->pColumnDataAgg[0];
  int32_t type = pInput->pData[0]->info.type;

  SSpreadInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));

  if (pInput->colDataAggIsSet) {
    numOfElems = pInput->numOfRows - pAgg->numOfNull;
    if (numOfElems == 0) {
      goto _spread_over;
    }
    double tmin = 0.0, tmax = 0.0;
    if (IS_SIGNED_NUMERIC_TYPE(type)) {
      tmin = (double)GET_INT64_VAL(&pAgg->min);
      tmax = (double)GET_INT64_VAL(&pAgg->max);
    } else if (IS_FLOAT_TYPE(type)) {
      tmin = GET_DOUBLE_VAL(&pAgg->min);
      tmax = GET_DOUBLE_VAL(&pAgg->max);
    } else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
      tmin = (double)GET_UINT64_VAL(&pAgg->min);
      tmax = (double)GET_UINT64_VAL(&pAgg->max);
    }

    if (GET_DOUBLE_VAL(&pInfo->min) > tmin) {
      SET_DOUBLE_VAL(&pInfo->min, tmin);
    }

    if (GET_DOUBLE_VAL(&pInfo->max) < tmax) {
      SET_DOUBLE_VAL(&pInfo->max, tmax);
    }

  } else {  // computing based on the true data block
    SColumnInfoData* pCol = pInput->pData[0];

    int32_t start     = pInput->startRowIndex;
    int32_t numOfRows = pInput->numOfRows;

    // check the valid data one by one
    for (int32_t i = start; i < pInput->numOfRows + start; ++i) {
      if (colDataIsNull_f(pCol->nullbitmap, i)) {
        continue;
      }

      char *data = colDataGetData(pCol, i);

      double v = 0;
      GET_TYPED_DATA(v, double, type, data);
      if (v < GET_DOUBLE_VAL(&pInfo->min)) {
        SET_DOUBLE_VAL(&pInfo->min, v);
      }

      if (v > GET_DOUBLE_VAL(&pInfo->max)) {
        SET_DOUBLE_VAL(&pInfo->max, v);
      }

      numOfElems += 1;
    }
  }

_spread_over:
  // data in the check operation are all null, not output
  SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1);
  if (numOfElems > 0) {
    pInfo->hasResult = true;
  }

  return TSDB_CODE_SUCCESS;
}

int32_t spreadFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
  SSpreadInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
  if (pInfo->hasResult == true) {
    SET_DOUBLE_VAL(&pInfo->result, pInfo->max - pInfo->min);
  }
  return functionFinalize(pCtx, pBlock);
}
2443 2444 2445 2446 2447 2448

bool getHistogramFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(SHistoFuncInfo) + HISTOGRAM_MAX_BINS_NUM * sizeof(SHistoFuncBin);
  return true;
}

2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464 2465 2466 2467 2468 2469 2470 2471 2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489 2490 2491 2492 2493
static int8_t getHistogramBinType(char *binTypeStr) {
  int8_t binType;
  if (strcasecmp(binTypeStr, "user_input") == 0) {
    binType = USER_INPUT_BIN;
  } else if (strcasecmp(binTypeStr, "linear_bin") == 0) {
    binType = LINEAR_BIN;
  } else if (strcasecmp(binTypeStr, "log_bin") == 0) {
    binType = LOG_BIN;
  } else {
    binType = UNKNOWN_BIN;
  }

  return binType;
}

static bool getHistogramBinDesc(SHistoFuncInfo *pInfo, char *binDescStr, int8_t binType, bool normalized) {
  cJSON*  binDesc = cJSON_Parse(binDescStr);
  int32_t numOfBins;
  double* intervals;
  if (cJSON_IsObject(binDesc)) { /* linaer/log bins */
    int32_t numOfParams = cJSON_GetArraySize(binDesc);
    int32_t startIndex;
    if (numOfParams != 4) {
      return false;
    }

    cJSON* start    = cJSON_GetObjectItem(binDesc, "start");
    cJSON* factor   = cJSON_GetObjectItem(binDesc, "factor");
    cJSON* width    = cJSON_GetObjectItem(binDesc, "width");
    cJSON* count    = cJSON_GetObjectItem(binDesc, "count");
    cJSON* infinity = cJSON_GetObjectItem(binDesc, "infinity");

    if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) {
      return false;
    }

    if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000
      return false;
    }

    if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) ||
        (factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) {
      return false;
    }

2494
    int32_t counter = (int32_t)count->valueint;
2495 2496 2497 2498 2499 2500 2501 2502 2503 2504 2505 2506 2507 2508 2509 2510 2511 2512 2513 2514 2515 2516 2517 2518 2519 2520 2521 2522 2523 2524 2525 2526 2527 2528 2529 2530 2531 2532 2533 2534 2535 2536 2537 2538 2539 2540 2541 2542 2543 2544 2545 2546 2547 2548 2549 2550 2551 2552 2553
    if (infinity->valueint == false) {
      startIndex = 0;
      numOfBins = counter + 1;
    } else {
      startIndex = 1;
      numOfBins = counter + 3;
    }

    intervals = taosMemoryCalloc(numOfBins, sizeof(double));
    if (cJSON_IsNumber(width) && factor == NULL && binType == LINEAR_BIN) {
      // linear bin process
      if (width->valuedouble == 0) {
        taosMemoryFree(intervals);
        return false;
      }
      for (int i = 0; i < counter + 1; ++i) {
        intervals[startIndex] = start->valuedouble + i * width->valuedouble;
        if (isinf(intervals[startIndex])) {
          taosMemoryFree(intervals);
          return false;
        }
        startIndex++;
      }
    } else if (cJSON_IsNumber(factor) && width == NULL && binType == LOG_BIN) {
      // log bin process
      if (start->valuedouble == 0) {
        taosMemoryFree(intervals);
        return false;
      }
      if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) {
        taosMemoryFree(intervals);
        return false;
      }
      for (int i = 0; i < counter + 1; ++i) {
        intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0);
        if (isinf(intervals[startIndex])) {
          taosMemoryFree(intervals);
          return false;
        }
        startIndex++;
      }
    } else {
      taosMemoryFree(intervals);
      return false;
    }

    if (infinity->valueint == true) {
      intervals[0] = -INFINITY;
      intervals[numOfBins - 1] = INFINITY;
      // in case of desc bin orders, -inf/inf should be swapped
      ASSERT(numOfBins >= 4);
      if (intervals[1] > intervals[numOfBins - 2]) {
        TSWAP(intervals[0], intervals[numOfBins - 1]);
      }
    }
  } else if (cJSON_IsArray(binDesc)) { /* user input bins */
    if (binType != USER_INPUT_BIN) {
      return false;
    }
2554
    numOfBins = cJSON_GetArraySize(binDesc);
2555 2556 2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567 2568 2569 2570 2571 2572 2573 2574 2575 2576 2577 2578
    intervals = taosMemoryCalloc(numOfBins, sizeof(double));
    cJSON* bin = binDesc->child;
    if (bin == NULL) {
      taosMemoryFree(intervals);
      return false;
    }
    int i = 0;
    while (bin) {
      intervals[i] = bin->valuedouble;
      if (!cJSON_IsNumber(bin)) {
        taosMemoryFree(intervals);
        return false;
      }
      if (i != 0 && intervals[i] <= intervals[i - 1]) {
        taosMemoryFree(intervals);
        return false;
      }
      bin = bin->next;
      i++;
    }
  } else {
    return false;
  }

2579
  pInfo->numOfBins  = numOfBins - 1;
2580
  pInfo->normalized = normalized;
2581
  for (int32_t i = 0; i < pInfo->numOfBins; ++i) {
2582 2583 2584 2585 2586 2587 2588 2589 2590 2591
    pInfo->bins[i].lower = intervals[i] < intervals[i + 1] ? intervals[i] : intervals[i + 1];
    pInfo->bins[i].upper = intervals[i + 1] > intervals[i] ? intervals[i + 1] : intervals[i];
    pInfo->bins[i].count = 0;
  }

  taosMemoryFree(intervals);
  return true;
}

bool histogramFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo *pResultInfo) {
2592 2593 2594 2595
  if (!functionSetup(pCtx, pResultInfo)) {
    return false;
  }

2596
  SHistoFuncInfo *pInfo = GET_ROWCELL_INTERBUF(pResultInfo);
2597 2598 2599
  pInfo->numOfBins = 0;
  pInfo->totalCount = 0;
  pInfo->normalized = 0;
2600

2601 2602 2603 2604 2605 2606
  int8_t binType = getHistogramBinType(varDataVal(pCtx->param[1].param.pz));
  if (binType == UNKNOWN_BIN) {
    return false;
  }
  char* binDesc = varDataVal(pCtx->param[2].param.pz);
  int64_t normalized = pCtx->param[3].param.i;
2607 2608 2609
  if (normalized != 0 && normalized != 1) {
    return false;
  }
2610 2611 2612
  if (!getHistogramBinDesc(pInfo, binDesc, binType, (bool)normalized)) {
    return false;
  }
2613 2614 2615 2616 2617

  return true;
}

int32_t histogramFunction(SqlFunctionCtx *pCtx) {
2618 2619 2620 2621 2622 2623 2624 2625 2626 2627
  SHistoFuncInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));

  SInputColumnInfoData* pInput = &pCtx->input;
  SColumnInfoData*      pCol = pInput->pData[0];

  int32_t type = pInput->pData[0]->info.type;

  int32_t start = pInput->startRowIndex;
  int32_t numOfRows = pInput->numOfRows;

2628
  int32_t numOfElems = 0;
2629 2630 2631 2632 2633
  for (int32_t i = start; i < numOfRows + start; ++i) {
    if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
      continue;
    }

2634 2635
    numOfElems++;

2636 2637 2638 2639 2640 2641 2642
    char* data = colDataGetData(pCol, i);
    double v;
    GET_TYPED_DATA(v, double, type, data);

    for (int32_t k = 0; k < pInfo->numOfBins; ++k) {
      if (v > pInfo->bins[k].lower && v <= pInfo->bins[k].upper) {
        pInfo->bins[k].count++;
2643
        pInfo->totalCount++;
2644 2645 2646
        break;
      }
    }
2647 2648 2649

  }

2650
  SET_VAL(GET_RES_INFO(pCtx), numOfElems, pInfo->numOfBins);
2651 2652 2653 2654
  return TSDB_CODE_SUCCESS;
}

int32_t histogramFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
2655
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
2656
  SHistoFuncInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
2657 2658 2659 2660 2661
  int32_t        slotId = pCtx->pExpr->base.resSchema.slotId;
  SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);

  int32_t currentRow = pBlock->info.rows;

2662 2663 2664 2665 2666 2667 2668 2669 2670 2671
  if (pInfo->normalized) {
    for (int32_t k = 0; k < pResInfo->numOfRes; ++k) {
      if(pInfo->totalCount != 0) {
        pInfo->bins[k].percentage = pInfo->bins[k].count / (double)pInfo->totalCount;
      } else {
        pInfo->bins[k].percentage = 0;
      }
    }
  }

2672
  for (int32_t i = 0; i < pResInfo->numOfRes; ++i) {
2673
    int32_t len;
2674
    char buf[512] = {0};
2675
    if (!pInfo->normalized) {
2676
      len = sprintf(varDataVal(buf), "{\"lower_bin\":%g, \"upper_bin\":%g, \"count\":%"PRId64"}",
2677 2678
                   pInfo->bins[i].lower, pInfo->bins[i].upper, pInfo->bins[i].count);
    } else {
2679
      len = sprintf(varDataVal(buf), "{\"lower_bin\":%g, \"upper_bin\":%g, \"count\":%lf}",
2680 2681 2682 2683 2684 2685 2686
                   pInfo->bins[i].lower, pInfo->bins[i].upper, pInfo->bins[i].percentage);
    }
    varDataSetLen(buf, len);
    colDataAppend(pCol, currentRow, buf, false);
    currentRow++;
  }

2687
  return pResInfo->numOfRes;
2688
}
2689 2690 2691 2692 2693 2694 2695 2696 2697 2698 2699 2700 2701 2702 2703 2704 2705 2706 2707 2708 2709 2710 2711 2712 2713 2714 2715 2716

bool getStateFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(SStateInfo);
  return true;
}

static int8_t getStateOpType(char *opStr) {
  int8_t opType;
  if (strcasecmp(opStr, "LT") == 0) {
    opType = STATE_OPER_LT;
  } else if (strcasecmp(opStr, "GT") == 0) {
    opType = STATE_OPER_GT;
  } else if (strcasecmp(opStr, "LE") == 0) {
    opType = STATE_OPER_LE;
  } else if (strcasecmp(opStr, "GE") == 0) {
    opType = STATE_OPER_GE;
  } else if (strcasecmp(opStr, "NE") == 0) {
    opType = STATE_OPER_NE;
  } else if (strcasecmp(opStr, "EQ") == 0) {
    opType = STATE_OPER_EQ;
  } else {
    opType = STATE_OPER_INVALID;
  }

  return opType;
}

#define GET_STATE_VAL(param) \
G
Ganlin Zhao 已提交
2717
  ((param.nType == TSDB_DATA_TYPE_BIGINT) ? (param.i) : (param.d))
2718 2719

#define STATE_COMP(_op, _lval, _param)  \
G
Ganlin Zhao 已提交
2720 2721 2722 2723 2724 2725 2726 2727 2728 2729 2730 2731 2732 2733 2734 2735 2736 2737 2738 2739 2740 2741 2742 2743 2744 2745 2746
  STATE_COMP_IMPL(_op, _lval, GET_STATE_VAL(_param))

#define STATE_COMP_IMPL(_op, _lval, _rval)  \
  do {                                      \
    switch(_op) {                           \
      case STATE_OPER_LT:                   \
        return ((_lval) < (_rval));         \
        break;                              \
      case STATE_OPER_GT:                   \
        return ((_lval) > (_rval));         \
        break;                              \
      case STATE_OPER_LE:                   \
        return ((_lval) <= (_rval));        \
        break;                              \
      case STATE_OPER_GE:                   \
        return ((_lval) >= (_rval));        \
        break;                              \
      case STATE_OPER_NE:                   \
        return ((_lval) != (_rval));        \
        break;                              \
      case STATE_OPER_EQ:                   \
        return ((_lval) == (_rval));        \
        break;                              \
      default:                              \
        break;                              \
    }                                       \
  } while (0)
2747 2748 2749 2750 2751 2752 2753 2754 2755 2756 2757 2758 2759 2760 2761 2762 2763 2764 2765 2766 2767 2768 2769 2770 2771 2772 2773 2774 2775 2776 2777 2778 2779 2780 2781 2782 2783 2784 2785 2786 2787 2788 2789 2790 2791 2792 2793 2794 2795 2796 2797 2798 2799 2800 2801 2802 2803 2804 2805 2806 2807 2808 2809 2810 2811 2812 2813 2814 2815 2816 2817 2818 2819 2820

static bool checkStateOp(int8_t op, SColumnInfoData* pCol, int32_t index, SVariant param) {
  char* data = colDataGetData(pCol, index);
  switch(pCol->info.type) {
    case TSDB_DATA_TYPE_TINYINT: {
      int8_t v = *(int8_t *)data;
      STATE_COMP(op, v, param);
      break;
    }
    case TSDB_DATA_TYPE_UTINYINT: {
      uint8_t v = *(uint8_t *)data;
      STATE_COMP(op, v, param);
      break;
    }
    case TSDB_DATA_TYPE_SMALLINT: {
      int16_t v = *(int16_t *)data;
      STATE_COMP(op, v, param);
      break;
    }
    case TSDB_DATA_TYPE_USMALLINT: {
      uint16_t v = *(uint16_t *)data;
      STATE_COMP(op, v, param);
      break;
    }
    case TSDB_DATA_TYPE_INT: {
      int32_t v = *(int32_t *)data;
      STATE_COMP(op, v, param);
      break;
    }
    case TSDB_DATA_TYPE_UINT: {
      uint32_t v = *(uint32_t *)data;
      STATE_COMP(op, v, param);
      break;
    }
    case TSDB_DATA_TYPE_BIGINT: {
      int64_t v = *(int64_t *)data;
      STATE_COMP(op, v, param);
      break;
    }
    case TSDB_DATA_TYPE_UBIGINT: {
      uint64_t v = *(uint64_t *)data;
      STATE_COMP(op, v, param);
      break;
    }
    case TSDB_DATA_TYPE_FLOAT: {
      float v = *(float *)data;
      STATE_COMP(op, v, param);
      break;
    }
    case TSDB_DATA_TYPE_DOUBLE: {
      double v = *(double *)data;
      STATE_COMP(op, v, param);
      break;
    }
    default: {
      ASSERT(0);
    }
  }
  return false;
}

int32_t stateCountFunction(SqlFunctionCtx* pCtx) {
  SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
  SStateInfo*          pInfo = GET_ROWCELL_INTERBUF(pResInfo);

  SInputColumnInfoData* pInput = &pCtx->input;

  SColumnInfoData* pInputCol = pInput->pData[0];
  SColumnInfoData* pTsOutput = pCtx->pTsOutput;

  int32_t numOfElems = 0;
  SColumnInfoData* pOutput = (SColumnInfoData*)pCtx->pOutput;

  int8_t op = getStateOpType(varDataVal(pCtx->param[1].param.pz));
2821 2822 2823 2824
  if (STATE_OPER_INVALID == op) {
    return 0;
  }

2825 2826 2827 2828 2829 2830 2831 2832 2833 2834 2835 2836 2837 2838 2839 2840 2841 2842 2843
  for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; i += 1) {
    numOfElems++;
    if (colDataIsNull_f(pInputCol->nullbitmap, i)) {
      colDataAppendNULL(pOutput, i);
      continue;
    }

    bool ret = checkStateOp(op, pInputCol, i, pCtx->param[2].param);
    int64_t output = -1;
    if (ret) {
      output = ++pInfo->count;
    } else {
      pInfo->count = 0;
    }
    colDataAppend(pOutput, i, (char *)&output, false);
  }

  return numOfElems;
}