/*

 *

 * 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 "decimal.h"

#include "tdataformat.h"

#include "wideInteger.h"

typedef enum DecimalInternalType {

  DECIMAL_64 = 0,

  DECIMAL_128 = 1,

} DecimalInternalType;

typedef enum DecimalRoundType {

  ROUND_TYPE_CEIL,

  ROUND_TYPE_FLOOR,

  ROUND_TYPE_TRUNC,

  ROUND_TYPE_HALF_ROUND_UP,

} DecimalRoundType;

#define DECIMAL_GET_INTERNAL_TYPE(dataType) ((dataType) == TSDB_DATA_TYPE_DECIMAL ? DECIMAL_128 : DECIMAL_64)

#define DECIMAL_GET_WORD_NUM(decimalInternalType) \

  ((decimalInternalType) == DECIMAL_64 ? DECIMAL_WORD_NUM(Decimal64) : DECIMAL_WORD_NUM(Decimal128))

static SDecimalOps* getDecimalOpsImp(DecimalInternalType t);

#define DECIMAL_MIN_ADJUSTED_SCALE 6

static Decimal64 SCALE_MULTIPLIER_64[TSDB_DECIMAL64_MAX_PRECISION + 1] = {1LL,

                                                                          10LL,

                                                                          100LL,

                                                                          1000LL,

                                                                          10000LL,

                                                                          100000LL,

                                                                          1000000LL,

                                                                          10000000LL,

                                                                          100000000LL,

                                                                          1000000000LL,

                                                                          10000000000LL,

                                                                          100000000000LL,

                                                                          1000000000000LL,

                                                                          10000000000000LL,

                                                                          100000000000000LL,

                                                                          1000000000000000LL,

                                                                          10000000000000000LL,

                                                                          100000000000000000LL,

                                                                          1000000000000000000LL};

typedef struct DecimalVar {

  DecimalInternalType type;

  int8_t              precision;

  int8_t              scale;

  int32_t             exponent;

  int8_t              sign;

  DecimalType*        pDec;

  int32_t             weight;

} DecimalVar;

static uint8_t maxPrecision(DecimalInternalType type) {

  switch (type) {

    case DECIMAL_64:

      return TSDB_DECIMAL64_MAX_PRECISION;

    case DECIMAL_128:

      return TSDB_DECIMAL128_MAX_PRECISION;

    default:

      return 0;

  }

}

static const uint8_t typeConvertDecimalPrec[] = {

    0, 1, 3, 5, 10, 19, TSDB_DECIMAL128_MAX_PRECISION, TSDB_DECIMAL_MAX_PRECISION, 0, 19, 10, 3, 5, 10, 20, 0,

    0, 0, 0, 0, 0,  0};

int32_t decimalGetRetType(const SDataType* pLeftT, const SDataType* pRightT, EOperatorType opType,

                          SDataType* pOutType) {

  if (pLeftT->type == TSDB_DATA_TYPE_JSON || pRightT->type == TSDB_DATA_TYPE_JSON ||

      pLeftT->type == TSDB_DATA_TYPE_VARBINARY || pRightT->type == TSDB_DATA_TYPE_VARBINARY)

    return TSDB_CODE_TSC_INVALID_OPERATION;

  if ((pLeftT->type >= TSDB_DATA_TYPE_BLOB && pLeftT->type <= TSDB_DATA_TYPE_GEOMETRY) ||

      (pRightT->type >= TSDB_DATA_TYPE_BLOB && pRightT->type <= TSDB_DATA_TYPE_GEOMETRY)) {

    return TSDB_CODE_TSC_INVALID_OPERATION;

  }

  if (IS_FLOAT_TYPE(pLeftT->type) || IS_FLOAT_TYPE(pRightT->type) || IS_VAR_DATA_TYPE(pLeftT->type) ||

      IS_VAR_DATA_TYPE(pRightT->type)) {

    pOutType->type = TSDB_DATA_TYPE_DOUBLE;

    pOutType->bytes = tDataTypes[TSDB_DATA_TYPE_DOUBLE].bytes;

    return 0;

  }

  if (IS_NULL_TYPE(pLeftT->type) || IS_NULL_TYPE(pRightT->type)) {

    pOutType->type = TSDB_DATA_TYPE_NULL;

    pOutType->bytes = tDataTypes[TSDB_DATA_TYPE_NULL].bytes;

    return 0;

  }

  uint8_t p1 = pLeftT->precision, s1 = pLeftT->scale, p2 = pRightT->precision, s2 = pRightT->scale;

  if (!IS_DECIMAL_TYPE(pLeftT->type)) {

    p1 = typeConvertDecimalPrec[pLeftT->type];

    s1 = 0;

  }

  if (!IS_DECIMAL_TYPE(pRightT->type)) {

    p2 = typeConvertDecimalPrec[pRightT->type];

    s2 = 0;

  }

  switch (opType) {

    case OP_TYPE_ADD:

    case OP_TYPE_SUB:

      pOutType->scale = TMAX(s1, s2);

      pOutType->precision = TMAX(p1 - s1, p2 - s2) + pOutType->scale + 1;

      break;

    case OP_TYPE_MULTI:

      pOutType->scale = s1 + s2;

      pOutType->precision = p1 + p2 + 1;

      break;

    case OP_TYPE_DIV:

      pOutType->scale = TMAX(s1 + p2 + 1, DECIMAL_MIN_ADJUSTED_SCALE);

      pOutType->precision = p1 - s1 + s2 + pOutType->scale;

      break;

    case OP_TYPE_REM:

      pOutType->scale = TMAX(s1, s2);

      pOutType->precision = TMIN(p1 - s1, p2 - s2) + pOutType->scale;

      break;

    default:

      return TSDB_CODE_TSC_INVALID_OPERATION;

  }

  if (pOutType->precision > TSDB_DECIMAL_MAX_PRECISION) {

    int8_t minScale = TMIN(DECIMAL_MIN_ADJUSTED_SCALE, pOutType->scale);

    int8_t delta = pOutType->precision - TSDB_DECIMAL_MAX_PRECISION;

    pOutType->precision = TSDB_DECIMAL_MAX_PRECISION;

    pOutType->scale = TMAX(minScale, (int8_t)(pOutType->scale) - delta);

  }

  pOutType->type = TSDB_DATA_TYPE_DECIMAL;

  pOutType->bytes = tDataTypes[pOutType->type].bytes;

  return 0;

}

int32_t calcCurPrec(int32_t prec, int32_t places, int32_t exp, int32_t weight, int32_t firstValidScale) {

  if (exp == 0) return prec + places;

  if (exp < 0) {

    if (weight + exp >= 0) return prec + places;

    return prec + places - exp - weight;

  }

  if (weight > 0) return prec + places;

  return prec + places - TMIN(firstValidScale - 1, exp);

}

int32_t calcActualWeight(int32_t prec, int32_t scale, int32_t exp, int32_t weight, int32_t firstValidScale) {

  if (exp == 0) return prec - scale;

  if (exp < 0) {

    if (weight + exp >= 0) return weight + exp;

    return 0;

  }

  if (weight > 0) return weight + exp;

  if (firstValidScale == 0) return 0;

  return TMAX(0, exp - firstValidScale);

}

static int32_t decimalVarFromStr(const char* str, int32_t len, DecimalVar* result) {

  int32_t code = 0, pos = 0;

  int32_t expectPrecision = result->precision;

  int32_t expectScale = result->scale;

  result->precision = 0;

  result->scale = 0;

  result->exponent = 0;

  bool     leadingZeroes = true, afterPoint = false, rounded = false, stop = false;

  uint32_t places = 0;

  result->sign = 1;

  int32_t weight = 0;

  int32_t firstValidScale = 0;

  if (len == 0) return TSDB_CODE_INVALID_DATA_FMT;

  SDecimalOps* pOps = getDecimalOpsImp(result->type);

  // sign

  switch (str[pos]) {

    case '-':

      result->sign = -1;

    case '+':

      pos++;

    default:

      break;

  }

  int32_t pos2 = pos;

  while (pos2 < len) {

    if (str[pos2] == 'e' || str[pos2] == 'E') {

      int32_t expSign = 1;

      int32_t exp = 0;

      int32_t expPos = pos2 + 1;

      if (expPos < len && (str[expPos] == '+' || str[expPos] == '-')) {

        expSign = (str[expPos] == '-') ? -1 : 1;

        expPos++;

      }

      while (expPos < len && isdigit((unsigned char)str[expPos])) {

        int32_t digit = str[expPos] - '0';

        if (exp > (INT32_MAX - digit) / 10) {

          exp = INT32_MAX;

          break;

        }

        exp = exp * 10 + digit;

        expPos++;

      }

      result->exponent = (expSign > 0) ? exp : -exp;

      break;

    }

    pos2++;

  }

  for (; pos < len && !stop; ++pos) {

    switch (str[pos]) {

      case '.':

        weight = result->precision;

        afterPoint = true;

        leadingZeroes = false;

        break;

      case '0':

        if (leadingZeroes) break;

        if (afterPoint) {

          places++;

          break;

        }

      case '1':

      case '2':

      case '3':

      case '4':

      case '5':

      case '6':

      case '7':

      case '8':

      case '9': {

        leadingZeroes = false;

        ++places;

        if (firstValidScale == 0 && afterPoint) firstValidScale = places;

        int32_t   curPrec = calcCurPrec(result->precision, places, result->exponent, weight, firstValidScale);

        int32_t   scaleUp = 0;

        Decimal64 delta = {0};

        if (curPrec > maxPrecision(result->type)) {

          if (!afterPoint) return TSDB_CODE_DECIMAL_OVERFLOW;

          int32_t curScale = result->scale - result->exponent + places;

          if (rounded || curScale > expectScale + 1 /*scale already overflowed, no need do rounding*/ ||

              curPrec - 1 != maxPrecision(result->type) /* not the maxPrecision + 1 digit, no need do rounding*/ ||

              str[pos] < '5')

            break;

          // Do rounding for the maxPrecision + 1 digit.

          // Here we cannot directly add this digit into the results, because it may cause overflow.

          DECIMAL64_SET_VALUE(&delta, 1);

          scaleUp = places - 1;

          rounded = true;

        } else {

          scaleUp = places;

          DECIMAL64_SET_VALUE(&delta, str[pos] - '0');

        }

        result->precision += scaleUp;

        if (afterPoint) result->scale += scaleUp;

        while (scaleUp != 0) {

          int32_t curScale = TMIN(17, scaleUp);

          pOps->multiply(result->pDec, &SCALE_MULTIPLIER_64[curScale], DECIMAL_WORD_NUM(Decimal64));

          scaleUp -= curScale;

        }

        pOps->add(result->pDec, &delta, DECIMAL_WORD_NUM(Decimal64));

        places = 0;

      } break;

      case 'e':

      case 'E': {

          stop = true;

        } break;

      default:

        stop = true;

        break;

    }

  }

  result->weight = calcActualWeight(result->precision, result->scale, result->exponent, weight, firstValidScale);

  if (result->precision + result->scale > 0) result->scale -= result->exponent;

  if (result->sign < 0) {

    pOps->negate(result->pDec);

  }

  return code;

}

int32_t decimal64ToDataVal(const Decimal64* dec, SValue* pVal) {

  VALUE_SET_TRIVIAL_DATUM(pVal, DECIMAL64_GET_VALUE(dec));

  return TSDB_CODE_SUCCESS;

}

int32_t decimal128ToDataVal(Decimal128* dec, SValue* pVal) {

  void* pV = taosMemCalloc(1, sizeof(Decimal128));

  if (!pV) return terrno;

  memcpy(pV, dec, DECIMAL_WORD_NUM(Decimal128) * sizeof(DecimalWord));

  valueSetDatum(pVal, TSDB_DATA_TYPE_DECIMAL, pV, DECIMAL_WORD_NUM(Decimal128) * sizeof(DecimalWord));

  return TSDB_CODE_SUCCESS;

}

#define DECIMAL64_ONE  SCALE_MULTIPLIER_64[0]

#define DECIMAL64_GET_MAX(precision, pMax)                                \

  do {                                                                    \

    *(pMax) = SCALE_MULTIPLIER_64[precision];                             \

    decimal64Subtract(pMax, &DECIMAL64_ONE, DECIMAL_WORD_NUM(Decimal64)); \

  } while (0)

#define DECIMAL64_SIGN(pDec) (1 | (DECIMAL64_GET_VALUE(pDec) >> 63))

static void decimal64GetWhole(const DecimalType* pDec, int8_t scale, DecimalType* pWhole) {

  const SDecimalOps* pOps = getDecimalOps(TSDB_DATA_TYPE_DECIMAL64);

  DECIMAL64_CLONE(pWhole, pDec);

  Decimal64 scaleMul = SCALE_MULTIPLIER_64[scale];

  pOps->divide(pWhole, &scaleMul, 1, NULL);

  pOps->abs(pWhole);

}

static void decimal64GetFrac(const DecimalType* pDec, int8_t scale, DecimalType* pFrac) {

  const SDecimalOps* pOps = getDecimalOpsImp(DECIMAL_64);

  DECIMAL64_CLONE(pFrac, pDec);

  Decimal64 scaleMul = SCALE_MULTIPLIER_64[scale];

  pOps->mod(pFrac, &scaleMul, 1);

  pOps->abs(pFrac);

}

static void    decimal64Negate(DecimalType* pInt);

static void    decimal64Abs(DecimalType* pInt);

static void    decimal64Add(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static void    decimal64Subtract(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static void    decimal64Multiply(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static void    decimal64divide(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum,

                               DecimalType* pRemainder);

static void    decimal64Mod(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static bool    decimal64Lt(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static bool    decimal64Gt(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static bool    decimal64Eq(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static void    decimal64ScaleDown(Decimal64* pDec, uint8_t scaleDown, bool round);

static void    decimal64ScaleUp(Decimal64* pDec, uint8_t scaleUp);

static void    decimal64ScaleTo(Decimal64* pDec, uint8_t oldScale, uint8_t newScale);

int32_t        decimal64ToStr(const DecimalType* pInt, uint8_t scale, char* pBuf, int32_t bufLen);

static void decimal64RoundWithPositiveScale(Decimal64* pDec, uint8_t prec, int8_t scale, uint8_t toPrec,

                                            uint8_t toScale, DecimalRoundType roundType, bool* overflow);

static void    decimal128Negate(DecimalType* pInt);

static void    decimal128Abs(DecimalType* pWord);

static void    decimal128Add(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static void    decimal128Subtract(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static void    decimal128Multiply(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static void    decimal128Divide(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum,

                                DecimalType* pRemainder);

static void    decimal128Mod(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static bool    decimal128Lt(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static bool    decimal128Gt(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static bool    decimal128Eq(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum);

static void    decimal128ScaleTo(Decimal128* pDec, uint8_t oldScale, uint8_t newScale);

static void    decimal128ScaleDown(Decimal128* pDec, uint8_t scaleDown, bool round);

static void    decimal128ScaleUp(Decimal128* pDec, uint8_t scaleUp);

static int32_t decimal128CountLeadingBinaryZeros(const Decimal128* pDec);

static int32_t decimal128FromInt64(DecimalType* pDec, uint8_t prec, uint8_t scale, int64_t val);

static int32_t decimal128FromUint64(DecimalType* pDec, uint8_t prec, uint8_t scale, uint64_t val);

int32_t        decimal128ToStr(const DecimalType* pInt, uint8_t scale, char* pBuf, int32_t bufLen);

//

// rounding functions

static void    decimal128RoundWithPositiveScale(Decimal128* pDec, uint8_t prec, uint8_t scale, uint8_t toPrec,

                                                uint8_t toScale, DecimalRoundType roundType, bool* overflow);

static void    decimal128RoundWithNegativeScale(Decimal128* pDec, uint8_t prec, uint8_t scale, int8_t toScale,

                                                DecimalRoundType roundType, bool* overflow);

static void    decimal128ModifyScaleAndPrecision(Decimal128* pDec, uint8_t scale, uint8_t toPrec, int8_t toScale,

                                                 bool* overflow);

static int32_t decimal128CountRoundingDelta(const Decimal128* pDec, int8_t scale, int8_t toScale,

                                            DecimalRoundType roundType);

SDecimalOps decimal64Ops = {decimal64Negate,   decimal64Abs,    decimal64Add,  decimal64Subtract,

                            decimal64Multiply, decimal64divide, decimal64Mod,  decimal64Lt,

                            decimal64Gt,       decimal64Eq,     decimal64ToStr};

SDecimalOps decimal128Ops = {decimal128Negate,   decimal128Abs,    decimal128Add,  decimal128Subtract,

                             decimal128Multiply, decimal128Divide, decimal128Mod,  decimal128Lt,

                             decimal128Gt,       decimal128Eq,     decimal128ToStr};

static SDecimalOps* getDecimalOpsImp(DecimalInternalType t) {

  switch (t) {

    case DECIMAL_128:

      return &decimal128Ops;

    case DECIMAL_64:

      return &decimal64Ops;

    default:

      return NULL;

  }

}

const SDecimalOps* getDecimalOps(int8_t dataType) { return getDecimalOpsImp(DECIMAL_GET_INTERNAL_TYPE(dataType)); }

void makeDecimal64(Decimal64* pDec64, int64_t w) { DECIMAL64_SET_VALUE(pDec64, w); }

void decimal64Negate(DecimalType* pInt) {

  Decimal64* pDec = pInt;

  DECIMAL64_SET_VALUE(pDec, -DECIMAL64_GET_VALUE(pDec));

}

void decimal64Abs(DecimalType* pInt) {

  Decimal64* pDec = pInt;

  DECIMAL64_SET_VALUE(pDec, TABS(DECIMAL64_GET_VALUE(pDec)));

}

void decimal64Add(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal64*       pDecL = pLeft;

  const Decimal64* pDecR = pRight;

  DECIMAL64_SET_VALUE(pDecL, SAFE_INT64_ADD(DECIMAL64_GET_VALUE(pDecL), DECIMAL64_GET_VALUE(pDecR)));

}

void decimal64Subtract(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal64*       pDecL = pLeft;

  const Decimal64* pDecR = pRight;

  DECIMAL64_SET_VALUE(pDecL, SAFE_INT64_SUBTRACT(DECIMAL64_GET_VALUE(pDecL), DECIMAL64_GET_VALUE(pDecR)));

}

void decimal64Multiply(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal64* pDecL = pLeft;

  Decimal64  decR = *((Decimal64*)pRight);

  bool       sign = DECIMAL64_SIGN(pDecL) != DECIMAL64_SIGN(&decR);

  decimal64Abs(pLeft);

  decimal64Abs(&decR);

  uint64_t x = DECIMAL64_GET_VALUE(pDecL), y = DECIMAL64_GET_VALUE(&decR);

  x *= y;

  DECIMAL64_SET_VALUE(pDecL, x);

  if (sign) decimal64Negate(pDecL);

}

void decimal64divide(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum, DecimalType* pRemainder) {

  Decimal64* pDecL = pLeft;

  Decimal64  decR = *((Decimal64*)pRight);

  Decimal64* pDecRemainder = pRemainder;

  bool       sign = DECIMAL64_SIGN(pDecL) != DECIMAL64_SIGN(&decR);

  decimal64Abs(pDecL);

  decimal64Abs(&decR);

  uint64_t x = DECIMAL64_GET_VALUE(pDecL), y = DECIMAL64_GET_VALUE(&decR);

  uint64_t z = x;

  x /= y;

  DECIMAL64_SET_VALUE(pDecL, x);

  if (sign) decimal64Negate(pDecL);

  if (pDecRemainder) {

    z %= y;

    DECIMAL64_SET_VALUE(pDecRemainder, z);

  }

}

void decimal64Mod(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal64  remainder = {0};

  Decimal64* pDec = pLeft;

  decimal64divide(pLeft, pRight, rightWordNum, &remainder);

  DECIMAL64_SET_VALUE(pDec, DECIMAL64_GET_VALUE(&remainder));

}

bool decimal64Lt(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  const Decimal64 *pDecL = pLeft, *pDecR = pRight;

  return DECIMAL64_GET_VALUE(pDecL) < DECIMAL64_GET_VALUE(pDecR);

}

bool decimal64Gt(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  return DECIMAL64_GET_VALUE((Decimal64*)pLeft) > DECIMAL64_GET_VALUE((Decimal64*)pRight);

}

bool decimal64Eq(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  return DECIMAL64_GET_VALUE((Decimal64*)pLeft) == DECIMAL64_GET_VALUE(((Decimal64*)pRight));

}

int32_t decimal64ToStr(const DecimalType* pInt, uint8_t scale, char* pBuf, int32_t bufLen) {

  if (!pBuf) return TSDB_CODE_INVALID_PARA;

  Decimal whole = {0}, frac = {0};

  int32_t pos = 0;

  char    buf[64] = {0};

  if (DECIMAL64_SIGN((Decimal64*)pInt) == -1) {

    pos = snprintf(buf, sizeof(buf), "-");

  }

  decimal64GetWhole(pInt, scale, &whole);

  pos += snprintf(buf + pos, bufLen - pos, "%" PRId64, DECIMAL64_GET_VALUE(&whole));

  if (scale > 0) {

    decimal64GetFrac(pInt, scale, &frac);

    if (DECIMAL64_GET_VALUE(&frac) != 0 || DECIMAL64_GET_VALUE(&whole) != 0) {

      TAOS_STRCAT(buf + pos, ".");

      pos += 1;

      // NOTE: never generate format string dynamically

      //       decimalTest has been passed.

      snprintf(buf + pos, bufLen - pos, "%0*" PRIu64, scale, DECIMAL64_GET_VALUE(&frac));

    }

  }

  TAOS_STRNCPY(pBuf, buf, bufLen);

  return 0;

}

// return 1 if positive or zero, else return -1

#define DECIMAL128_SIGN(pDec) (1 | (DECIMAL128_HIGH_WORD(pDec) >> 63))

static const Decimal128 SCALE_MULTIPLIER_128[TSDB_DECIMAL128_MAX_PRECISION + 1] = {

    DEFINE_DECIMAL128(1LL, 0),

    DEFINE_DECIMAL128(10LL, 0),

    DEFINE_DECIMAL128(100LL, 0),

    DEFINE_DECIMAL128(1000LL, 0),

    DEFINE_DECIMAL128(10000LL, 0),

    DEFINE_DECIMAL128(100000LL, 0),

    DEFINE_DECIMAL128(1000000LL, 0),

    DEFINE_DECIMAL128(10000000LL, 0),

    DEFINE_DECIMAL128(100000000LL, 0),

    DEFINE_DECIMAL128(1000000000LL, 0),

    DEFINE_DECIMAL128(10000000000LL, 0),

    DEFINE_DECIMAL128(100000000000LL, 0),

    DEFINE_DECIMAL128(1000000000000LL, 0),

    DEFINE_DECIMAL128(10000000000000LL, 0),

    DEFINE_DECIMAL128(100000000000000LL, 0),

    DEFINE_DECIMAL128(1000000000000000LL, 0),

    DEFINE_DECIMAL128(10000000000000000LL, 0),

    DEFINE_DECIMAL128(100000000000000000LL, 0),

    DEFINE_DECIMAL128(1000000000000000000LL, 0),

    DEFINE_DECIMAL128(10000000000000000000ULL, 0LL),

    DEFINE_DECIMAL128(7766279631452241920ULL, 5LL),

    DEFINE_DECIMAL128(3875820019684212736ULL, 54LL),

    DEFINE_DECIMAL128(1864712049423024128ULL, 542LL),

    DEFINE_DECIMAL128(200376420520689664ULL, 5421LL),

    DEFINE_DECIMAL128(2003764205206896640ULL, 54210LL),

    DEFINE_DECIMAL128(1590897978359414784ULL, 542101LL),

    DEFINE_DECIMAL128(15908979783594147840ULL, 5421010LL),

    DEFINE_DECIMAL128(11515845246265065472ULL, 54210108LL),

    DEFINE_DECIMAL128(4477988020393345024ULL, 542101086LL),

    DEFINE_DECIMAL128(7886392056514347008ULL, 5421010862LL),

    DEFINE_DECIMAL128(5076944270305263616ULL, 54210108624LL),

    DEFINE_DECIMAL128(13875954555633532928ULL, 542101086242LL),

    DEFINE_DECIMAL128(9632337040368467968ULL, 5421010862427LL),

    DEFINE_DECIMAL128(4089650035136921600ULL, 54210108624275LL),

    DEFINE_DECIMAL128(4003012203950112768ULL, 542101086242752LL),

    DEFINE_DECIMAL128(3136633892082024448ULL, 5421010862427522LL),

    DEFINE_DECIMAL128(12919594847110692864ULL, 54210108624275221LL),

    DEFINE_DECIMAL128(68739955140067328ULL, 542101086242752217LL),

    DEFINE_DECIMAL128(687399551400673280ULL, 5421010862427522170LL),

};

static double getDoubleScaleMultiplier(uint8_t scale) {

  static double SCALE_MULTIPLIER_DOUBLE[TSDB_DECIMAL_MAX_PRECISION + 1] = {0};

  static bool   initialized = false;

  if (!initialized) {

    SCALE_MULTIPLIER_DOUBLE[0] = 1.0;

    for (int32_t idx = 1; idx <= TSDB_DECIMAL_MAX_PRECISION; ++idx) {

      SCALE_MULTIPLIER_DOUBLE[idx] = SCALE_MULTIPLIER_DOUBLE[idx - 1] * 10;

    }

    initialized = true;

  }

  return SCALE_MULTIPLIER_DOUBLE[scale];

};

#define DECIMAL128_ONE  SCALE_MULTIPLIER_128[0]

#define DECIMAL128_TEN  SCALE_MULTIPLIER_128[1]

// To calculate how many bits for integer X.

// eg. 999(3 digits) -> 1111100111(10 bits) -> bitsForNumDigits[3] = 10

static const int32_t bitsForNumDigits[] = {0,  4,  7,  10, 14,  17,  20,  24,  27,  30,  34,  37,  40,

                                           44, 47, 50, 54, 57,  60,  64,  67,  70,  74,  77,  80,  84,

                                           87, 90, 94, 97, 100, 103, 107, 110, 113, 117, 120, 123, 127};

#define DECIMAL128_GET_MAX(precision, pMax)                                  \

  do {                                                                       \

    *(pMax) = SCALE_MULTIPLIER_128[precision];                               \

    decimal128Subtract(pMax, &DECIMAL128_ONE, DECIMAL_WORD_NUM(Decimal128)); \

  } while (0)

void makeDecimal128(Decimal128* pDec128, int64_t hi, uint64_t low) {

  DECIMAL128_SET_HIGH_WORD(pDec128, hi);

  DECIMAL128_SET_LOW_WORD(pDec128, low);

}

static void makeDecimal128FromDecimal64(Decimal128* pTarget, Decimal64 decimal64) {

  bool negative = false;

  if (DECIMAL64_SIGN(&decimal64) == -1) {

    decimal64Negate(&decimal64);

    negative = true;

  }

  makeDecimal128(pTarget, 0, DECIMAL64_GET_VALUE(&decimal64));

  if (negative) decimal128Negate(pTarget);

}

static void decimal128Negate(DecimalType* pWord) {

  Decimal128* pDec = (Decimal128*)pWord;

  uint64_t    lo = ~DECIMAL128_LOW_WORD(pDec) + 1;

  int64_t     hi = ~DECIMAL128_HIGH_WORD(pDec);

  if (lo == 0) hi = SAFE_INT64_ADD(hi, 1);

  makeDecimal128(pDec, hi, lo);

}

static void decimal128Abs(DecimalType* pWord) {

  if (DECIMAL128_SIGN((Decimal128*)pWord) == -1) {

    decimal128Negate(pWord);

  }

}

#define DECIMAL128_CHECK_RIGHT_WORD_NUM(rightWordNum, pTarget, rightDec, pWord) \

  if (rightWordNum != DECIMAL_WORD_NUM(Decimal128)) {                                   \

    Decimal64 d64 = {0};                                                        \

    makeDecimal64(&d64, *(int64_t*)pWord);                                      \

    makeDecimal128FromDecimal64(&rightDec, d64);                                \

    pTarget = &rightDec;                                                        \

  }

static void decimal128Add(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal128 *pLeftDec = (Decimal128*)pLeft, *pRightDec = (Decimal128*)pRight;

  Decimal128  right = {0};

  DECIMAL128_CHECK_RIGHT_WORD_NUM(rightWordNum, pRightDec, right, pRight);

  int64_t  hi = SAFE_INT64_ADD(DECIMAL128_HIGH_WORD(pLeftDec), DECIMAL128_HIGH_WORD(pRightDec));

  uint64_t lo = DECIMAL128_LOW_WORD(pLeftDec) + DECIMAL128_LOW_WORD(pRightDec);

  hi = SAFE_INT64_ADD(hi, lo < DECIMAL128_LOW_WORD(pLeftDec));

  makeDecimal128(pLeftDec, hi, lo);

}

static void decimal128Subtract(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal128 *pLeftDec = (Decimal128*)pLeft, *pRightDec = (Decimal128*)pRight;

  Decimal128  right = {0};

  DECIMAL128_CHECK_RIGHT_WORD_NUM(rightWordNum, pRightDec, right, pRight);

  int64_t  hi = SAFE_INT64_SUBTRACT(DECIMAL128_HIGH_WORD(pLeftDec), DECIMAL128_HIGH_WORD(pRightDec));

  uint64_t lo = DECIMAL128_LOW_WORD(pLeftDec) - DECIMAL128_LOW_WORD(pRightDec);

  hi = SAFE_INT64_SUBTRACT(hi, lo > DECIMAL128_LOW_WORD(pLeftDec));

  makeDecimal128(pLeftDec, hi, lo);

}

static void decimal128Multiply(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal128 *pLeftDec = (Decimal128*)pLeft, *pRightDec = (Decimal128*)pRight;

  Decimal128  right = {0};

  DECIMAL128_CHECK_RIGHT_WORD_NUM(rightWordNum, pRightDec, right, pRight);

  bool       negate = DECIMAL128_SIGN(pLeftDec) != DECIMAL128_SIGN(pRightDec);

  Decimal128 x = *pLeftDec, y = *pRightDec;

  decimal128Abs(&x);

  decimal128Abs(&y);

  UInt128 res = {0}, tmp = {0};

  makeUInt128(&res, DECIMAL128_HIGH_WORD(&x), DECIMAL128_LOW_WORD(&x));

  makeUInt128(&tmp, DECIMAL128_HIGH_WORD(&y), DECIMAL128_LOW_WORD(&y));

  uInt128Multiply(&res, &tmp);

  makeDecimal128(pLeftDec, uInt128Hi(&res), uInt128Lo(&res));

  if (negate) decimal128Negate(pLeftDec);

}

static bool decimal128Lt(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal128 *pLeftDec = (Decimal128*)pLeft, *pRightDec = (Decimal128*)pRight;

  Decimal128  right = {0};

  DECIMAL128_CHECK_RIGHT_WORD_NUM(rightWordNum, pRightDec, right, pRight);

  return DECIMAL128_HIGH_WORD(pLeftDec) < DECIMAL128_HIGH_WORD(pRightDec) ||

         (DECIMAL128_HIGH_WORD(pLeftDec) == DECIMAL128_HIGH_WORD(pRightDec) &&

          DECIMAL128_LOW_WORD(pLeftDec) < DECIMAL128_LOW_WORD(pRightDec));

}

static void decimal128Divide(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum,

                             DecimalType* pRemainder) {

  Decimal128 *pLeftDec = (Decimal128*)pLeft, *pRightDec = (Decimal128*)pRight, *pRemainderDec = (Decimal128*)pRemainder;

  Decimal128  right = {0};

  DECIMAL128_CHECK_RIGHT_WORD_NUM(rightWordNum, pRightDec, right, pRight);

  bool leftNegate = DECIMAL128_SIGN(pLeftDec) == -1, rightNegate = DECIMAL128_SIGN(pRightDec) == -1;

  UInt128    a = {0}, b = {0}, c = {0}, d = {0};

  Decimal128 x = *pLeftDec, y = *pRightDec;

  decimal128Abs(&x);

  decimal128Abs(&y);

  makeUInt128(&a, DECIMAL128_HIGH_WORD(&x), DECIMAL128_LOW_WORD(&x));

  makeUInt128(&d, DECIMAL128_HIGH_WORD(&x), DECIMAL128_LOW_WORD(&x));

  makeUInt128(&b, DECIMAL128_HIGH_WORD(&y), DECIMAL128_LOW_WORD(&y));

  uInt128Divide(&a, &b);

  uInt128Mod(&d, &b);

  makeDecimal128(pLeftDec, uInt128Hi(&a), uInt128Lo(&a));

  if (pRemainder) makeDecimal128(pRemainderDec, uInt128Hi(&d), uInt128Lo(&d));

  if (leftNegate != rightNegate) decimal128Negate(pLeftDec);

  if (leftNegate && pRemainder) decimal128Negate(pRemainderDec);

}

static void decimal128Mod(DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal128 pLeftDec = *(Decimal128*)pLeft, *pRightDec = (Decimal128*)pRight, right = {0};

  DECIMAL128_CHECK_RIGHT_WORD_NUM(rightWordNum, pRightDec, right, pRight);

  decimal128Divide(&pLeftDec, pRightDec, DECIMAL_WORD_NUM(Decimal128),

                   pLeft);

}

static bool decimal128Gt(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal128 *pLeftDec = (Decimal128*)pLeft, *pRightDec = (Decimal128*)pRight;

  Decimal128  right = {0};

  DECIMAL128_CHECK_RIGHT_WORD_NUM(rightWordNum, pRightDec, right, pRight);

  return decimal128Lt(pRightDec, pLeftDec, DECIMAL_WORD_NUM(Decimal128));

}

static bool decimal128Eq(const DecimalType* pLeft, const DecimalType* pRight, uint8_t rightWordNum) {

  Decimal128 *pLeftDec = (Decimal128*)pLeft, *pRightDec = (Decimal128*)pRight;

  Decimal128  right = {0};

  DECIMAL128_CHECK_RIGHT_WORD_NUM(rightWordNum, pRightDec, right, pRight);

  return DECIMAL128_HIGH_WORD(pLeftDec) == DECIMAL128_HIGH_WORD(pRightDec) &&

         DECIMAL128_LOW_WORD(pLeftDec) == DECIMAL128_LOW_WORD(pRightDec);

}

#define DIGIT_NUM_ONCE 18

static void extractDecimal128Digits(const Decimal128* pDec, uint64_t* digits, int32_t* digitNum) {

  UInt128 a = {0};

  UInt128 b = {0};

  *digitNum = 0;

  makeUInt128(&a, DECIMAL128_HIGH_WORD(pDec), DECIMAL128_LOW_WORD(pDec));

  while (!uInt128Eq(&a, &uInt128Zero)) {

    uint64_t hi = uInt128Hi(&a);

    uint64_t lo = uInt128Lo(&a);

    uint64_t hiQuotient = hi / k1e18;

    uint64_t hiRemainder = hi % k1e18;

    makeUInt128(&b, hiRemainder, lo);

    uInt128Divide(&b, &uInt128_1e18);

    uint64_t loQuotient = uInt128Lo(&b);

    makeUInt128(&b, hiRemainder, lo);

    uInt128Mod(&b, &uInt128_1e18);

    uint64_t loRemainder = uInt128Lo(&b);

    makeUInt128(&a, hiQuotient, loQuotient);

    digits[(*digitNum)++] = loRemainder;

  }

}

int32_t decimal128ToStr(const DecimalType* pInt, uint8_t scale, char* pBuf, int32_t bufLen) {

  if (!pBuf) return TSDB_CODE_INVALID_PARA;

  const Decimal128* pDec = (const Decimal128*)pInt;

  bool              negative = DECIMAL128_SIGN(pDec) == -1;

  uint64_t          segments[3] = {0};

  int32_t           digitNum = 0;

  char              buf[64] = {0}, buf2[64] = {0};

  int32_t           len = 0;

  if (negative) {

    Decimal128 copy = {0};

    makeDecimal128(&copy, DECIMAL128_HIGH_WORD(pDec), DECIMAL128_LOW_WORD(pDec));

    decimal128Abs(&copy);

    extractDecimal128Digits(&copy, segments, &digitNum);

    TAOS_STRNCAT(buf2, "-", 2);

  } else {

    extractDecimal128Digits(pDec, segments, &digitNum);

  }

  if (digitNum == 0) {

    TAOS_STRNCPY(pBuf, "0", bufLen);

    return 0;

  }

  for (int32_t i = digitNum - 1; i >= 0; --i) {

    len += snprintf(buf + len, 64 - len, i == digitNum - 1 ? "%" PRIu64 : "%018" PRIu64, segments[i]);

  }

  int32_t wholeLen = len - scale;

  if (wholeLen > 0) {

    TAOS_STRNCAT(buf2, buf, wholeLen);

  } else {

    TAOS_STRNCAT(buf2, "0", 2);

  }

  if (scale > 0) {

    static const char *format = "0000000000000000000000000000000000000000";

    TAOS_STRNCAT(buf2, ".", 2);

    if (wholeLen < 0) TAOS_STRNCAT(buf2, format, TABS(wholeLen));

    TAOS_STRNCAT(buf2, buf + TMAX(0, wholeLen), scale);

  }

  TAOS_STRNCPY(pBuf, buf2, bufLen);

  return 0;

}

int32_t decimalToStr(const DecimalType* pDec, int8_t dataType, int8_t precision, int8_t scale, char* pBuf,

                     int32_t bufLen) {

  DecimalInternalType iType = DECIMAL_GET_INTERNAL_TYPE(dataType);

  switch (iType) {

    case DECIMAL_64:

      return decimal64ToStr(pDec, scale, pBuf, bufLen);

    case DECIMAL_128:

      return decimal128ToStr(pDec, scale, pBuf, bufLen);

    default:

      break;

  }

  return TSDB_CODE_INVALID_PARA;

}

static int32_t decimalAddLargePositive(Decimal* pX, const SDataType* pXT, const Decimal* pY, const SDataType* pYT,

                                    const SDataType* pOT) {

  Decimal wholeX = *pX, wholeY = *pY, fracX = {0}, fracY = {0};

  decimal128Divide(&wholeX, &SCALE_MULTIPLIER_128[pXT->scale], DECIMAL_WORD_NUM(Decimal), &fracX);

  decimal128Divide(&wholeY, &SCALE_MULTIPLIER_128[pYT->scale], DECIMAL_WORD_NUM(Decimal), &fracY);

  uint8_t maxScale = TMAX(pXT->scale, pYT->scale);

  decimal128ScaleUp(&fracX, maxScale - pXT->scale);

  decimal128ScaleUp(&fracY, maxScale - pYT->scale);

  Decimal pMultiplier = SCALE_MULTIPLIER_128[maxScale];

  Decimal right = fracX;

  Decimal carry = {0};

  decimal128Subtract(&pMultiplier, &fracY, DECIMAL_WORD_NUM(Decimal));

  if (!decimal128Gt(&pMultiplier, &fracX, DECIMAL_WORD_NUM(Decimal))) {

    decimal128Subtract(&right, &pMultiplier, DECIMAL_WORD_NUM(Decimal));

    makeDecimal128(&carry, 0, 1);

  } else {

    decimal128Add(&right, &fracY, DECIMAL_WORD_NUM(Decimal));

  }

  decimal128ScaleDown(&right, maxScale - pOT->scale, true);

  if (decimal128AddCheckOverflow(&wholeX, &wholeY, DECIMAL_WORD_NUM(Decimal))) {

    return TSDB_CODE_DECIMAL_OVERFLOW;

  }

  decimal128Add(&wholeX, &wholeY, DECIMAL_WORD_NUM(Decimal));

  if (decimal128AddCheckOverflow(&wholeX, &carry, DECIMAL_WORD_NUM(Decimal))) {

    return TSDB_CODE_DECIMAL_OVERFLOW;

  }

  decimal128Add(&wholeX, &carry, DECIMAL_WORD_NUM(Decimal));

  decimal128Multiply(&wholeX, &SCALE_MULTIPLIER_128[pOT->scale], DECIMAL_WORD_NUM(Decimal));

  decimal128Add(&wholeX, &right, DECIMAL_WORD_NUM(Decimal));

  *pX = wholeX;

  return 0;

}

static void decimalAddLargeNegative(Decimal* pX, const SDataType* pXT, const Decimal* pY, const SDataType* pYT,

                                    const SDataType* pOT) {

  Decimal wholeX = *pX, wholeY = *pY, fracX = {0}, fracY = {0};

  decimal128Divide(&wholeX, &SCALE_MULTIPLIER_128[pXT->scale], DECIMAL_WORD_NUM(Decimal), &fracX);

  decimal128Divide(&wholeY, &SCALE_MULTIPLIER_128[pYT->scale], DECIMAL_WORD_NUM(Decimal), &fracY);

  uint8_t maxScale = TMAX(pXT->scale, pYT->scale);

  decimal128ScaleUp(&fracX, maxScale - pXT->scale);

  decimal128ScaleUp(&fracY, maxScale - pYT->scale);

  decimal128Add(&wholeX, &wholeY, DECIMAL_WORD_NUM(Decimal));

  decimal128Add(&fracX, &fracY, DECIMAL_WORD_NUM(Decimal));

  if (DECIMAL128_SIGN(&wholeX) == -1 && decimal128Gt(&fracX, &DECIMAL128_ZERO, DECIMAL_WORD_NUM(Decimal128))) {

    decimal128Add(&wholeX, &DECIMAL128_ONE, DECIMAL_WORD_NUM(Decimal));

    decimal128Subtract(&fracX, &SCALE_MULTIPLIER_128[maxScale], DECIMAL_WORD_NUM(Decimal));

  } else if (decimal128Gt(&wholeX, &DECIMAL128_ZERO, DECIMAL_WORD_NUM(Decimal128)) && DECIMAL128_SIGN(&fracX) == -1) {

    decimal128Subtract(&wholeX, &DECIMAL128_ONE, DECIMAL_WORD_NUM(Decimal));

    decimal128Add(&fracX, &SCALE_MULTIPLIER_128[maxScale], DECIMAL_WORD_NUM(Decimal));

  }

  decimal128ScaleDown(&fracX, maxScale - pOT->scale, true);

  decimal128Multiply(&wholeX, &SCALE_MULTIPLIER_128[pOT->scale], DECIMAL_WORD_NUM(Decimal));

  decimal128Add(&wholeX, &fracX, DECIMAL_WORD_NUM(Decimal));

  *pX = wholeX;

}

static int32_t decimalAdd(Decimal* pX, const SDataType* pXT, const Decimal* pY, const SDataType* pYT,

                       const SDataType* pOT) {

  int32_t code = 0;

  if (pOT->precision < TSDB_DECIMAL_MAX_PRECISION) {

    uint8_t maxScale = TMAX(pXT->scale, pYT->scale);

    Decimal tmpY = *pY;

    decimal128ScaleTo(pX, pXT->scale, maxScale);

    decimal128ScaleTo(&tmpY, pYT->scale, maxScale);

    decimal128Add(pX, &tmpY, DECIMAL_WORD_NUM(Decimal));

  } else {

    int8_t signX = DECIMAL128_SIGN(pX), signY = DECIMAL128_SIGN(pY);

    if (signX == 1 && signY == 1) {

      code = decimalAddLargePositive(pX, pXT, pY, pYT, pOT);

    } else if (signX == -1 && signY == -1) {

      decimal128Negate(pX);