#3660

Committed 15 Mar 2025 09:06AM UTC coverage: 62.039% (-1.3%) from 63.314%

Build # #3660

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(stream): support stream processing for virtual tables (#30144)

* enh: add client processing

* enh: add mnode vtables processing

* enh: add mnode vtable processing

* enh: add normal child vtable support

* fix: compile issues

* fix: compile issues

* fix: create stream issues

* fix: multi stream scan issue

* fix: remove debug info

* fix: agg task and task level issues

* fix: correct task output type

* fix: split vtablescan from agg

* fix: memory leak issues

* fix: add limitations

* Update 09-error-code.md

* Update 09-error-code.md

* fix: remove usless case

* feat(stream): extract original table data in source scan task

Implemented functionality in the source task to extract data
corresponding to the virtual table from the original table using WAL.
The extracted data is then sent to the downstream merge task for further
processing.

* feat(stream): multi-way merge using loser tree in virtual merge task

Implemented multi-way merge in the merge task using a loser tree to
combine data from multiple original table into a single virtual table.
The merged virtual table data is then pushed downstream for further
processing.  Introduced memory limit handling during the merge process
with configurable behavior when the memory limit is reached.

* fix(test): remove useless cases

---------

Co-authored-by: dapan1121 <wpan@taosdata.com>
Co-authored-by: Pan Wei <72057773+dapan1121@users.noreply.github.com>

Run Details

154078 of 317582 branches covered (48.52%)

Branch coverage included in aggregate %.

313 of 2391 new or added lines in 34 files covered. (13.09%)

26134 existing lines in 205 files now uncovered.

240261 of 318051 relevant lines covered (75.54%)

16655189.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.03

/source/common/src/ttypes.c

/*
 * 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/>.
 */

#define _DEFAULT_SOURCE
#include "ttypes.h"
#include "tcompression.h"

const int32_t TYPE_BYTES[TSDB_DATA_TYPE_MAX] = {
    2,                      // TSDB_DATA_TYPE_NULL
    CHAR_BYTES,              // TSDB_DATA_TYPE_BOOL
    CHAR_BYTES,              // TSDB_DATA_TYPE_TINYINT
    SHORT_BYTES,             // TSDB_DATA_TYPE_SMALLINT
    INT_BYTES,               // TSDB_DATA_TYPE_INT
    sizeof(int64_t),         // TSDB_DATA_TYPE_BIGINT
    FLOAT_BYTES,             // TSDB_DATA_TYPE_FLOAT
    DOUBLE_BYTES,            // TSDB_DATA_TYPE_DOUBLE
    sizeof(VarDataOffsetT),  // TSDB_DATA_TYPE_BINARY
    sizeof(TSKEY),           // TSDB_DATA_TYPE_TIMESTAMP
    sizeof(VarDataOffsetT),  // TSDB_DATA_TYPE_NCHAR
    CHAR_BYTES,              // TSDB_DATA_TYPE_UTINYINT
    SHORT_BYTES,             // TSDB_DATA_TYPE_USMALLINT
    INT_BYTES,               // TSDB_DATA_TYPE_UINT
    sizeof(uint64_t),        // TSDB_DATA_TYPE_UBIGINT
    TSDB_MAX_JSON_TAG_LEN,   // TSDB_DATA_TYPE_JSON
    sizeof(VarDataOffsetT),  // TSDB_DATA_TYPE_VARBINARY
    DECIMAL128_BYTES,        // TSDB_DATA_TYPE_DECIMAL: placeholder, not implemented
    TSDB_MAX_TAGS_LEN,       // TSDB_DATA_TYPE_BLOB: placeholder, not implemented
    TSDB_MAX_TAGS_LEN,       // TSDB_DATA_TYPE_MEDIUMBLOB: placeholder, not implemented
    sizeof(VarDataOffsetT),  // TSDB_DATA_TYPE_GEOMETRY
    DECIMAL64_BYTES,         // TSDB_DATA_TYPE_DECIMAL64
};

tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX] = {
    {TSDB_DATA_TYPE_NULL, 6, 2, "NOTYPE", 0, 0, NULL, NULL},
    {TSDB_DATA_TYPE_BOOL, 4, CHAR_BYTES, "BOOL", false, true, tsCompressBool, tsDecompressBool},
    {TSDB_DATA_TYPE_TINYINT, 7, CHAR_BYTES, "TINYINT", INT8_MIN, INT8_MAX, tsCompressTinyint, tsDecompressTinyint},
    {TSDB_DATA_TYPE_SMALLINT, 8, SHORT_BYTES, "SMALLINT", INT16_MIN, INT16_MAX, tsCompressSmallint,
     tsDecompressSmallint},
    {TSDB_DATA_TYPE_INT, 3, INT_BYTES, "INT", INT32_MIN, INT32_MAX, tsCompressInt, tsDecompressInt},
    {TSDB_DATA_TYPE_BIGINT, 6, LONG_BYTES, "BIGINT", INT64_MIN, INT64_MAX, tsCompressBigint, tsDecompressBigint},
    {TSDB_DATA_TYPE_FLOAT, 5, FLOAT_BYTES, "FLOAT", 0, 0, tsCompressFloat, tsDecompressFloat},
    {TSDB_DATA_TYPE_DOUBLE, 6, DOUBLE_BYTES, "DOUBLE", 0, 0, tsCompressDouble, tsDecompressDouble},
    {TSDB_DATA_TYPE_VARCHAR, 6, 1, "VARCHAR", 0, 0, tsCompressString, tsDecompressString},
    {TSDB_DATA_TYPE_TIMESTAMP, 9, LONG_BYTES, "TIMESTAMP", INT64_MIN, INT64_MAX, tsCompressTimestamp,
     tsDecompressTimestamp},
    {TSDB_DATA_TYPE_NCHAR, 5, 1, "NCHAR", 0, 0, tsCompressString, tsDecompressString},
    {TSDB_DATA_TYPE_UTINYINT, 16, CHAR_BYTES, "TINYINT UNSIGNED", 0, UINT8_MAX, tsCompressTinyint, tsDecompressTinyint},
    {TSDB_DATA_TYPE_USMALLINT, 17, SHORT_BYTES, "SMALLINT UNSIGNED", 0, UINT16_MAX, tsCompressSmallint,
     tsDecompressSmallint},
    {TSDB_DATA_TYPE_UINT, 12, INT_BYTES, "INT UNSIGNED", 0, UINT32_MAX, tsCompressInt, tsDecompressInt},
    {TSDB_DATA_TYPE_UBIGINT, 15, LONG_BYTES, "BIGINT UNSIGNED", 0, UINT64_MAX, tsCompressBigint, tsDecompressBigint},
    {TSDB_DATA_TYPE_JSON, 4, TSDB_MAX_JSON_TAG_LEN, "JSON", 0, 0, tsCompressString, tsDecompressString},
    {TSDB_DATA_TYPE_VARBINARY, 9, 1, "VARBINARY", 0, 0, tsCompressString,
     tsDecompressString},                                                        // placeholder, not implemented
    {TSDB_DATA_TYPE_DECIMAL, 7, DECIMAL128_BYTES, "DECIMAL", 0, 0, NULL, NULL},  // placeholder, not implemented
    {TSDB_DATA_TYPE_BLOB, 4, 1, "BLOB", 0, 0, NULL, NULL},                       // placeholder, not implemented
    {TSDB_DATA_TYPE_MEDIUMBLOB, 10, 1, "MEDIUMBLOB", 0, 0, NULL, NULL},          // placeholder, not implemented
    {TSDB_DATA_TYPE_GEOMETRY, 8, 1, "GEOMETRY", 0, 0, tsCompressString, tsDecompressString},
    {TSDB_DATA_TYPE_DECIMAL64, 7, DECIMAL64_BYTES, "DECIMAL", 0, 0, NULL, NULL},
};

tDataTypeCompress tDataCompress[TSDB_DATA_TYPE_MAX] = {
    {TSDB_DATA_TYPE_NULL, 6, 1, "NOTYPE", 0, 0, NULL, NULL},
    {TSDB_DATA_TYPE_BOOL, 4, CHAR_BYTES, "BOOL", false, true, tsCompressBool2, tsDecompressBool2},
    {TSDB_DATA_TYPE_TINYINT, 7, CHAR_BYTES, "TINYINT", INT8_MIN, INT8_MAX, tsCompressTinyint2, tsDecompressTinyint2},
    {TSDB_DATA_TYPE_SMALLINT, 8, SHORT_BYTES, "SMALLINT", INT16_MIN, INT16_MAX, tsCompressSmallint2,
     tsDecompressSmallint2},
    {TSDB_DATA_TYPE_INT, 3, INT_BYTES, "INT", INT32_MIN, INT32_MAX, tsCompressInt2, tsDecompressInt2},
    {TSDB_DATA_TYPE_BIGINT, 6, LONG_BYTES, "BIGINT", INT64_MIN, INT64_MAX, tsCompressBigint2, tsDecompressBigint2},
    {TSDB_DATA_TYPE_FLOAT, 5, FLOAT_BYTES, "FLOAT", 0, 0, tsCompressFloat2, tsDecompressFloat2},
    {TSDB_DATA_TYPE_DOUBLE, 6, DOUBLE_BYTES, "DOUBLE", 0, 0, tsCompressDouble2, tsDecompressDouble2},
    {TSDB_DATA_TYPE_VARCHAR, 6, 1, "VARCHAR", 0, 0, tsCompressString2, tsDecompressString2},
    {TSDB_DATA_TYPE_TIMESTAMP, 9, LONG_BYTES, "TIMESTAMP", INT64_MIN, INT64_MAX, tsCompressTimestamp2,
     tsDecompressTimestamp2},
    {TSDB_DATA_TYPE_NCHAR, 5, 1, "NCHAR", 0, 0, tsCompressString2, tsDecompressString2},
    {TSDB_DATA_TYPE_UTINYINT, 16, CHAR_BYTES, "TINYINT UNSIGNED", 0, UINT8_MAX, tsCompressTinyint2,
     tsDecompressTinyint2},
    {TSDB_DATA_TYPE_USMALLINT, 17, SHORT_BYTES, "SMALLINT UNSIGNED", 0, UINT16_MAX, tsCompressSmallint2,
     tsDecompressSmallint2},
    {TSDB_DATA_TYPE_UINT, 12, INT_BYTES, "INT UNSIGNED", 0, UINT32_MAX, tsCompressInt2, tsDecompressInt2},
    {TSDB_DATA_TYPE_UBIGINT, 15, LONG_BYTES, "BIGINT UNSIGNED", 0, UINT64_MAX, tsCompressBigint2, tsDecompressBigint2},
    {TSDB_DATA_TYPE_JSON, 4, TSDB_MAX_JSON_TAG_LEN, "JSON", 0, 0, tsCompressString2, tsDecompressString2},
    {TSDB_DATA_TYPE_VARBINARY, 9, 1, "VARBINARY", 0, 0, tsCompressString2,
     tsDecompressString2},                                               // placeholder, not implemented
    {TSDB_DATA_TYPE_DECIMAL, 7, DECIMAL128_BYTES, "DECIMAL", 0, 0, tsCompressDecimal128, tsDecompressDecimal128},
    {TSDB_DATA_TYPE_BLOB, 4, 1, "BLOB", 0, 0, NULL, NULL},               // placeholder, not implemented
    {TSDB_DATA_TYPE_MEDIUMBLOB, 10, 1, "MEDIUMBLOB", 0, 0, NULL, NULL},  // placeholder, not implemented
    {TSDB_DATA_TYPE_GEOMETRY, 8, 1, "GEOMETRY", 0, 0, tsCompressString2, tsDecompressString2},
    {TSDB_DATA_TYPE_DECIMAL64, 9, DECIMAL64_BYTES, "DECIMAL64", 0, 0, tsCompressDecimal64, tsDecompressDecimal64},

};

static float  floatMin = -FLT_MAX, floatMax = FLT_MAX;
static double doubleMin = -DBL_MAX, doubleMax = DBL_MAX;

FORCE_INLINE void *getDataMin(int32_t type, void *value) {
  switch (type) {
    case TSDB_DATA_TYPE_FLOAT:
      *(float *)value = floatMin;
      break;
    case TSDB_DATA_TYPE_DOUBLE:
      *(double *)value = doubleMin;
      break;
    default:
      *(int64_t *)value = tDataTypes[type].minValue;
      break;
  }

  return value;
}

FORCE_INLINE void *getDataMax(int32_t type, void *value) {
  switch (type) {
    case TSDB_DATA_TYPE_FLOAT:
      *(float *)value = floatMax;
      break;
    case TSDB_DATA_TYPE_DOUBLE:
      *(double *)value = doubleMax;
      break;
    default:
      *(int64_t *)value = tDataTypes[type].maxValue;
      break;
  }

  return value;
}

bool isValidDataType(int32_t type) { return type >= TSDB_DATA_TYPE_NULL && type < TSDB_DATA_TYPE_MAX; }

#define POINTER_SHIFT(p, b) ((void *)((char *)(p) + (b)))

void assignVal(char *val, const char *src, int32_t len, int32_t type) {
  switch (type) {
    case TSDB_DATA_TYPE_BOOL:
    case TSDB_DATA_TYPE_TINYINT:
    case TSDB_DATA_TYPE_UTINYINT:
      *((int8_t *)val) = GET_INT8_VAL(src);
      break;
    case TSDB_DATA_TYPE_SMALLINT:
    case TSDB_DATA_TYPE_USMALLINT:
      *((int16_t *)val) = GET_INT16_VAL(src);
      break;
    case TSDB_DATA_TYPE_INT:
    case TSDB_DATA_TYPE_UINT:
      *((int32_t *)val) = GET_INT32_VAL(src);
      break;

    case TSDB_DATA_TYPE_FLOAT:
      SET_FLOAT_VAL(val, GET_FLOAT_VAL(src));
      break;
    case TSDB_DATA_TYPE_DOUBLE:
      SET_DOUBLE_VAL(val, GET_DOUBLE_VAL(src));
      break;
    case TSDB_DATA_TYPE_BIGINT:
    case TSDB_DATA_TYPE_UBIGINT:
    case TSDB_DATA_TYPE_TIMESTAMP:
      *((int64_t *)val) = GET_INT64_VAL(src);
      break;
    case TSDB_DATA_TYPE_BINARY:
    case TSDB_DATA_TYPE_VARBINARY:
    case TSDB_DATA_TYPE_GEOMETRY:
      varDataCopy(val, src);
      break;
    case TSDB_DATA_TYPE_NCHAR:
      varDataCopy(val, src);
      break;
    default: {
      if (len > 0) {
        (void)memcpy(val, src, len);
      }

      break;
    }
  }
}

int32_t operateVal(void *dst, void *s1, void *s2, int32_t optr, int32_t type) {
  if (optr == OP_TYPE_ADD) {
    switch (type) {
      case TSDB_DATA_TYPE_TINYINT:
        *((int8_t *)dst) = GET_INT8_VAL(s1) + GET_INT8_VAL(s2);
        break;
      case TSDB_DATA_TYPE_UTINYINT:
        *((uint8_t *)dst) = GET_UINT8_VAL(s1) + GET_UINT8_VAL(s2);
        break;
      case TSDB_DATA_TYPE_SMALLINT:
        *((int16_t *)dst) = GET_INT16_VAL(s1) + GET_INT16_VAL(s2);
        break;
      case TSDB_DATA_TYPE_USMALLINT:
        *((uint16_t *)dst) = GET_UINT16_VAL(s1) + GET_UINT16_VAL(s2);
        break;
      case TSDB_DATA_TYPE_INT:
        *((int32_t *)dst) = GET_INT32_VAL(s1) + GET_INT32_VAL(s2);
        break;
      case TSDB_DATA_TYPE_UINT:
        *((uint32_t *)dst) = GET_UINT32_VAL(s1) + GET_UINT32_VAL(s2);
        break;
      case TSDB_DATA_TYPE_BIGINT:
        *((int64_t *)dst) = GET_INT64_VAL(s1) + GET_INT64_VAL(s2);
        break;
      case TSDB_DATA_TYPE_UBIGINT:
        *((uint64_t *)dst) = GET_UINT64_VAL(s1) + GET_UINT64_VAL(s2);
        break;
      case TSDB_DATA_TYPE_TIMESTAMP:
        *((int64_t *)dst) = GET_INT64_VAL(s1) + GET_INT64_VAL(s2);
        break;
      case TSDB_DATA_TYPE_FLOAT:
        SET_FLOAT_VAL(dst, GET_FLOAT_VAL(s1) + GET_FLOAT_VAL(s2));
        break;
      case TSDB_DATA_TYPE_DOUBLE:
        SET_DOUBLE_VAL(dst, GET_DOUBLE_VAL(s1) + GET_DOUBLE_VAL(s2));
        break;
      default: {
        return -1;
      }
    }
  } else {
    return -1;
  }

  return 0;
}

uint8_t decimalTypeFromPrecision(uint8_t precision) {
  return precision > TSDB_DECIMAL64_MAX_PRECISION ? TSDB_DATA_TYPE_DECIMAL : TSDB_DATA_TYPE_DECIMAL64;
}

STypeMod decimalCalcTypeMod(uint8_t prec, uint8_t scale) {
  return ((STypeMod)prec << 8) + scale;
}

void decimalFromTypeMod(STypeMod typeMod, uint8_t* precision, uint8_t* scale) {
  if (precision) *precision = (uint8_t)((typeMod >> 8) & 0xFF);
  if (scale) *scale = (uint8_t)(typeMod & 0xFF);
}

STypeMod typeGetTypeModFromDataType(const SDataType* pDataType) {
  if (IS_DECIMAL_TYPE(pDataType->type)) return decimalCalcTypeMod(pDataType->precision, pDataType->scale);
  return 0;
}

STypeMod typeGetTypeMod(uint8_t type, uint8_t prec, uint8_t scale, int32_t bytes) {
  if (IS_DECIMAL_TYPE(type)) {
    return decimalCalcTypeMod(prec, scale);
  }
  return 0;
}

void fillTypeFromTypeMod(SDataType* pType, STypeMod mod) {
  if (IS_DECIMAL_TYPE(pType->type)) {
    decimalFromTypeMod(mod, &pType->precision, &pType->scale);
  }
}

void extractTypeFromTypeMod(uint8_t type, STypeMod typeMod, uint8_t *prec, uint8_t *scale, int32_t *bytes) {
  if (IS_DECIMAL_TYPE(type)) {
    decimalFromTypeMod(typeMod, prec, scale);
  } else {
    if (prec) *prec = 0;
    if (scale) *scale = 0;
  }
  if (bytes) *bytes = tDataTypes[type].bytes;
}

uint8_t getScaleFromTypeMod(int32_t type, STypeMod mod) {
  if (IS_DECIMAL_TYPE(type)) return (uint8_t)(mod & 0xFF);
  return 0;
}

// bytes, 0, prec, scale
void fillBytesForDecimalType(int32_t *pBytes, int32_t type, uint8_t precision, uint8_t scale) {
  *pBytes = 0;
  *pBytes = tDataTypes[type].bytes << 24;
  *pBytes |= (uint32_t)precision << 8;
  *pBytes |= scale;
}

void extractDecimalTypeInfoFromBytes(int32_t *pBytes, uint8_t *precision, uint8_t *scale) {
  *precision = (uint8_t)((*pBytes >> 8) & 0xFF);
  *scale = (uint8_t)(*pBytes & 0xFF);
  *pBytes >>= 24;
}

int32_t calcTypeBytesFromSchemaBytes(int32_t type, int32_t schemaBytes, bool isStmt) {
  if (isStmt) return schemaBytes;
  if (type == TSDB_DATA_TYPE_VARCHAR || type == TSDB_DATA_TYPE_VARBINARY || type == TSDB_DATA_TYPE_GEOMETRY) {
    return schemaBytes - VARSTR_HEADER_SIZE;
  } else if (type == TSDB_DATA_TYPE_NCHAR || type == TSDB_DATA_TYPE_JSON) {
    return (schemaBytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
  }
  return schemaBytes;
}

int32_t calcSchemaBytesFromTypeBytes(int32_t type, int32_t varTypeBytes, bool isStmt) {
  if (isStmt) return varTypeBytes;
  if (type == TSDB_DATA_TYPE_VARCHAR || type == TSDB_DATA_TYPE_VARBINARY || type == TSDB_DATA_TYPE_GEOMETRY) {
    return varTypeBytes + VARSTR_HEADER_SIZE;
  } else if (type == TSDB_DATA_TYPE_NCHAR || type == TSDB_DATA_TYPE_JSON) {
    return varTypeBytes * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
  }
  return varTypeBytes;
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#define _DEFAULT_SOURCE
17	#include "ttypes.h"
18	#include "tcompression.h"
19
20	const int32_t TYPE_BYTES[TSDB_DATA_TYPE_MAX] = {
21	2, // TSDB_DATA_TYPE_NULL
22	CHAR_BYTES, // TSDB_DATA_TYPE_BOOL
23	CHAR_BYTES, // TSDB_DATA_TYPE_TINYINT
24	SHORT_BYTES, // TSDB_DATA_TYPE_SMALLINT
25	INT_BYTES, // TSDB_DATA_TYPE_INT
26	sizeof(int64_t), // TSDB_DATA_TYPE_BIGINT
27	FLOAT_BYTES, // TSDB_DATA_TYPE_FLOAT
28	DOUBLE_BYTES, // TSDB_DATA_TYPE_DOUBLE
29	sizeof(VarDataOffsetT), // TSDB_DATA_TYPE_BINARY
30	sizeof(TSKEY), // TSDB_DATA_TYPE_TIMESTAMP
31	sizeof(VarDataOffsetT), // TSDB_DATA_TYPE_NCHAR
32	CHAR_BYTES, // TSDB_DATA_TYPE_UTINYINT
33	SHORT_BYTES, // TSDB_DATA_TYPE_USMALLINT
34	INT_BYTES, // TSDB_DATA_TYPE_UINT
35	sizeof(uint64_t), // TSDB_DATA_TYPE_UBIGINT
36	TSDB_MAX_JSON_TAG_LEN, // TSDB_DATA_TYPE_JSON
37	sizeof(VarDataOffsetT), // TSDB_DATA_TYPE_VARBINARY
38	DECIMAL128_BYTES, // TSDB_DATA_TYPE_DECIMAL: placeholder, not implemented
39	TSDB_MAX_TAGS_LEN, // TSDB_DATA_TYPE_BLOB: placeholder, not implemented
40	TSDB_MAX_TAGS_LEN, // TSDB_DATA_TYPE_MEDIUMBLOB: placeholder, not implemented
41	sizeof(VarDataOffsetT), // TSDB_DATA_TYPE_GEOMETRY
42	DECIMAL64_BYTES, // TSDB_DATA_TYPE_DECIMAL64
43	};
44
45	tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX] = {
46	{TSDB_DATA_TYPE_NULL, 6, 2, "NOTYPE", 0, 0, NULL, NULL},
47	{TSDB_DATA_TYPE_BOOL, 4, CHAR_BYTES, "BOOL", false, true, tsCompressBool, tsDecompressBool},
48	{TSDB_DATA_TYPE_TINYINT, 7, CHAR_BYTES, "TINYINT", INT8_MIN, INT8_MAX, tsCompressTinyint, tsDecompressTinyint},
49	{TSDB_DATA_TYPE_SMALLINT, 8, SHORT_BYTES, "SMALLINT", INT16_MIN, INT16_MAX, tsCompressSmallint,
50	tsDecompressSmallint},
51	{TSDB_DATA_TYPE_INT, 3, INT_BYTES, "INT", INT32_MIN, INT32_MAX, tsCompressInt, tsDecompressInt},
52	{TSDB_DATA_TYPE_BIGINT, 6, LONG_BYTES, "BIGINT", INT64_MIN, INT64_MAX, tsCompressBigint, tsDecompressBigint},
53	{TSDB_DATA_TYPE_FLOAT, 5, FLOAT_BYTES, "FLOAT", 0, 0, tsCompressFloat, tsDecompressFloat},
54	{TSDB_DATA_TYPE_DOUBLE, 6, DOUBLE_BYTES, "DOUBLE", 0, 0, tsCompressDouble, tsDecompressDouble},
55	{TSDB_DATA_TYPE_VARCHAR, 6, 1, "VARCHAR", 0, 0, tsCompressString, tsDecompressString},
56	{TSDB_DATA_TYPE_TIMESTAMP, 9, LONG_BYTES, "TIMESTAMP", INT64_MIN, INT64_MAX, tsCompressTimestamp,
57	tsDecompressTimestamp},
58	{TSDB_DATA_TYPE_NCHAR, 5, 1, "NCHAR", 0, 0, tsCompressString, tsDecompressString},
59	{TSDB_DATA_TYPE_UTINYINT, 16, CHAR_BYTES, "TINYINT UNSIGNED", 0, UINT8_MAX, tsCompressTinyint, tsDecompressTinyint},
60	{TSDB_DATA_TYPE_USMALLINT, 17, SHORT_BYTES, "SMALLINT UNSIGNED", 0, UINT16_MAX, tsCompressSmallint,
61	tsDecompressSmallint},
62	{TSDB_DATA_TYPE_UINT, 12, INT_BYTES, "INT UNSIGNED", 0, UINT32_MAX, tsCompressInt, tsDecompressInt},
63	{TSDB_DATA_TYPE_UBIGINT, 15, LONG_BYTES, "BIGINT UNSIGNED", 0, UINT64_MAX, tsCompressBigint, tsDecompressBigint},
64	{TSDB_DATA_TYPE_JSON, 4, TSDB_MAX_JSON_TAG_LEN, "JSON", 0, 0, tsCompressString, tsDecompressString},
65	{TSDB_DATA_TYPE_VARBINARY, 9, 1, "VARBINARY", 0, 0, tsCompressString,
66	tsDecompressString}, // placeholder, not implemented
67	{TSDB_DATA_TYPE_DECIMAL, 7, DECIMAL128_BYTES, "DECIMAL", 0, 0, NULL, NULL}, // placeholder, not implemented
68	{TSDB_DATA_TYPE_BLOB, 4, 1, "BLOB", 0, 0, NULL, NULL}, // placeholder, not implemented
69	{TSDB_DATA_TYPE_MEDIUMBLOB, 10, 1, "MEDIUMBLOB", 0, 0, NULL, NULL}, // placeholder, not implemented
70	{TSDB_DATA_TYPE_GEOMETRY, 8, 1, "GEOMETRY", 0, 0, tsCompressString, tsDecompressString},
71	{TSDB_DATA_TYPE_DECIMAL64, 7, DECIMAL64_BYTES, "DECIMAL", 0, 0, NULL, NULL},
72	};
73
74	tDataTypeCompress tDataCompress[TSDB_DATA_TYPE_MAX] = {
75	{TSDB_DATA_TYPE_NULL, 6, 1, "NOTYPE", 0, 0, NULL, NULL},
76	{TSDB_DATA_TYPE_BOOL, 4, CHAR_BYTES, "BOOL", false, true, tsCompressBool2, tsDecompressBool2},
77	{TSDB_DATA_TYPE_TINYINT, 7, CHAR_BYTES, "TINYINT", INT8_MIN, INT8_MAX, tsCompressTinyint2, tsDecompressTinyint2},
78	{TSDB_DATA_TYPE_SMALLINT, 8, SHORT_BYTES, "SMALLINT", INT16_MIN, INT16_MAX, tsCompressSmallint2,
79	tsDecompressSmallint2},
80	{TSDB_DATA_TYPE_INT, 3, INT_BYTES, "INT", INT32_MIN, INT32_MAX, tsCompressInt2, tsDecompressInt2},
81	{TSDB_DATA_TYPE_BIGINT, 6, LONG_BYTES, "BIGINT", INT64_MIN, INT64_MAX, tsCompressBigint2, tsDecompressBigint2},
82	{TSDB_DATA_TYPE_FLOAT, 5, FLOAT_BYTES, "FLOAT", 0, 0, tsCompressFloat2, tsDecompressFloat2},
83	{TSDB_DATA_TYPE_DOUBLE, 6, DOUBLE_BYTES, "DOUBLE", 0, 0, tsCompressDouble2, tsDecompressDouble2},
84	{TSDB_DATA_TYPE_VARCHAR, 6, 1, "VARCHAR", 0, 0, tsCompressString2, tsDecompressString2},
85	{TSDB_DATA_TYPE_TIMESTAMP, 9, LONG_BYTES, "TIMESTAMP", INT64_MIN, INT64_MAX, tsCompressTimestamp2,
86	tsDecompressTimestamp2},
87	{TSDB_DATA_TYPE_NCHAR, 5, 1, "NCHAR", 0, 0, tsCompressString2, tsDecompressString2},
88	{TSDB_DATA_TYPE_UTINYINT, 16, CHAR_BYTES, "TINYINT UNSIGNED", 0, UINT8_MAX, tsCompressTinyint2,
89	tsDecompressTinyint2},
90	{TSDB_DATA_TYPE_USMALLINT, 17, SHORT_BYTES, "SMALLINT UNSIGNED", 0, UINT16_MAX, tsCompressSmallint2,
91	tsDecompressSmallint2},
92	{TSDB_DATA_TYPE_UINT, 12, INT_BYTES, "INT UNSIGNED", 0, UINT32_MAX, tsCompressInt2, tsDecompressInt2},
93	{TSDB_DATA_TYPE_UBIGINT, 15, LONG_BYTES, "BIGINT UNSIGNED", 0, UINT64_MAX, tsCompressBigint2, tsDecompressBigint2},
94	{TSDB_DATA_TYPE_JSON, 4, TSDB_MAX_JSON_TAG_LEN, "JSON", 0, 0, tsCompressString2, tsDecompressString2},
95	{TSDB_DATA_TYPE_VARBINARY, 9, 1, "VARBINARY", 0, 0, tsCompressString2,
96	tsDecompressString2}, // placeholder, not implemented
97	{TSDB_DATA_TYPE_DECIMAL, 7, DECIMAL128_BYTES, "DECIMAL", 0, 0, tsCompressDecimal128, tsDecompressDecimal128},
98	{TSDB_DATA_TYPE_BLOB, 4, 1, "BLOB", 0, 0, NULL, NULL}, // placeholder, not implemented
99	{TSDB_DATA_TYPE_MEDIUMBLOB, 10, 1, "MEDIUMBLOB", 0, 0, NULL, NULL}, // placeholder, not implemented
100	{TSDB_DATA_TYPE_GEOMETRY, 8, 1, "GEOMETRY", 0, 0, tsCompressString2, tsDecompressString2},
101	{TSDB_DATA_TYPE_DECIMAL64, 9, DECIMAL64_BYTES, "DECIMAL64", 0, 0, tsCompressDecimal64, tsDecompressDecimal64},
102
103	};
104
105	static float floatMin = -FLT_MAX, floatMax = FLT_MAX;
106	static double doubleMin = -DBL_MAX, doubleMax = DBL_MAX;
107
108	FORCE_INLINE void getDataMin(int32_t type, void value) {	1,971,553✔
109	switch (type) {	1,971,553✔
110	case TSDB_DATA_TYPE_FLOAT:	15,043✔
111	(float )value = floatMin;	15,043✔
112	break;	15,043✔
113	case TSDB_DATA_TYPE_DOUBLE:	277,513✔
114	(double )value = doubleMin;	277,513✔
115	break;	277,513✔
116	default:	1,678,997✔
117	(int64_t )value = tDataTypes[type].minValue;	1,678,997✔
118	break;	1,678,997✔
119	}
120
121	return value;	1,971,553✔
122	}
123
124	FORCE_INLINE void getDataMax(int32_t type, void value) {	2,249,361✔
125	switch (type) {	2,249,361✔
126	case TSDB_DATA_TYPE_FLOAT:	14,723✔
127	(float )value = floatMax;	14,723✔
128	break;	14,723✔
129	case TSDB_DATA_TYPE_DOUBLE:	303,681✔
130	(double )value = doubleMax;	303,681✔
131	break;	303,681✔
132	default:	1,930,957✔
133	(int64_t )value = tDataTypes[type].maxValue;	1,930,957✔
134	break;	1,930,957✔
135	}
136
137	return value;	2,249,361✔
138	}
139
140	bool isValidDataType(int32_t type) { return type >= TSDB_DATA_TYPE_NULL && type < TSDB_DATA_TYPE_MAX; }	984,611!
141
142	#define POINTER_SHIFT(p, b) ((void )((char )(p) + (b)))
143
144	void assignVal(char val, const char src, int32_t len, int32_t type) {	16,869,743✔
145	switch (type) {	16,869,743!
146	case TSDB_DATA_TYPE_BOOL:	13,316,156✔
147	case TSDB_DATA_TYPE_TINYINT:
148	case TSDB_DATA_TYPE_UTINYINT:
149	((int8_t )val) = GET_INT8_VAL(src);	13,316,156✔
150	break;	13,316,156✔
151	case TSDB_DATA_TYPE_SMALLINT:	531,410✔
152	case TSDB_DATA_TYPE_USMALLINT:
153	((int16_t )val) = GET_INT16_VAL(src);	531,410✔
154	break;	531,410✔
155	case TSDB_DATA_TYPE_INT:	1,301,524✔
156	case TSDB_DATA_TYPE_UINT:
157	((int32_t )val) = GET_INT32_VAL(src);	1,301,524✔
158	break;	1,301,524✔
159
160	case TSDB_DATA_TYPE_FLOAT:	113,570✔
161	SET_FLOAT_VAL(val, GET_FLOAT_VAL(src));	113,570✔
162	break;	113,570✔
163	case TSDB_DATA_TYPE_DOUBLE:	440,592✔
164	SET_DOUBLE_VAL(val, GET_DOUBLE_VAL(src));	440,592✔
165	break;	440,592✔
166	case TSDB_DATA_TYPE_BIGINT:	799,783✔
167	case TSDB_DATA_TYPE_UBIGINT:
168	case TSDB_DATA_TYPE_TIMESTAMP:
169	((int64_t )val) = GET_INT64_VAL(src);	799,783✔
170	break;	799,783✔
171	case TSDB_DATA_TYPE_BINARY:	151,799✔
172	case TSDB_DATA_TYPE_VARBINARY:
173	case TSDB_DATA_TYPE_GEOMETRY:
174	varDataCopy(val, src);	151,799✔
175	break;	151,799✔
176	case TSDB_DATA_TYPE_NCHAR:	218,419✔
177	varDataCopy(val, src);	218,419✔
178	break;	218,419✔
UNCOV 179	default: {	×
UNCOV 180	if (len > 0) {	×
181	(void)memcpy(val, src, len);	×
182	}
183
UNCOV 184	break;	×
185	}
186	}
187	}	16,869,743✔
188
189	int32_t operateVal(void dst, void s1, void *s2, int32_t optr, int32_t type) {	174✔
190	if (optr == OP_TYPE_ADD) {	174!
191	switch (type) {	174!
192	case TSDB_DATA_TYPE_TINYINT:	×
193	((int8_t )dst) = GET_INT8_VAL(s1) + GET_INT8_VAL(s2);	×
194	break;	×
195	case TSDB_DATA_TYPE_UTINYINT:	×
196	((uint8_t )dst) = GET_UINT8_VAL(s1) + GET_UINT8_VAL(s2);	×
197	break;	×
198	case TSDB_DATA_TYPE_SMALLINT:	×
199	((int16_t )dst) = GET_INT16_VAL(s1) + GET_INT16_VAL(s2);	×
200	break;	×
201	case TSDB_DATA_TYPE_USMALLINT:	×
202	((uint16_t )dst) = GET_UINT16_VAL(s1) + GET_UINT16_VAL(s2);	×
203	break;	×
204	case TSDB_DATA_TYPE_INT:	×
205	((int32_t )dst) = GET_INT32_VAL(s1) + GET_INT32_VAL(s2);	×
206	break;	×
207	case TSDB_DATA_TYPE_UINT:	×
208	((uint32_t )dst) = GET_UINT32_VAL(s1) + GET_UINT32_VAL(s2);	×
209	break;	×
210	case TSDB_DATA_TYPE_BIGINT:	×
211	((int64_t )dst) = GET_INT64_VAL(s1) + GET_INT64_VAL(s2);	×
212	break;	×
UNCOV 213	case TSDB_DATA_TYPE_UBIGINT:	×
UNCOV 214	((uint64_t )dst) = GET_UINT64_VAL(s1) + GET_UINT64_VAL(s2);	×
UNCOV 215	break;	×
216	case TSDB_DATA_TYPE_TIMESTAMP:	174✔
217	((int64_t )dst) = GET_INT64_VAL(s1) + GET_INT64_VAL(s2);	174✔
218	break;	174✔
219	case TSDB_DATA_TYPE_FLOAT:	×
220	SET_FLOAT_VAL(dst, GET_FLOAT_VAL(s1) + GET_FLOAT_VAL(s2));	×
221	break;	×
222	case TSDB_DATA_TYPE_DOUBLE:	×
223	SET_DOUBLE_VAL(dst, GET_DOUBLE_VAL(s1) + GET_DOUBLE_VAL(s2));	×
UNCOV 224	break;	×
UNCOV 225	default: {	×
UNCOV 226	return -1;	×
227	}
228	}
229	} else {
UNCOV 230	return -1;	×
231	}
232
233	return 0;	174✔
234	}
235
236	uint8_t decimalTypeFromPrecision(uint8_t precision) {	701✔
237	return precision > TSDB_DECIMAL64_MAX_PRECISION ? TSDB_DATA_TYPE_DECIMAL : TSDB_DATA_TYPE_DECIMAL64;	701✔
238	}
239
240	STypeMod decimalCalcTypeMod(uint8_t prec, uint8_t scale) {	12,421,301✔
241	return ((STypeMod)prec << 8) + scale;	12,421,301✔
242	}
243
244	void decimalFromTypeMod(STypeMod typeMod, uint8_t* precision, uint8_t* scale) {	16,951,005✔
245	if (precision) *precision = (uint8_t)((typeMod >> 8) & 0xFF);	16,951,005!
246	if (scale) *scale = (uint8_t)(typeMod & 0xFF);	16,951,005!
247	}	16,951,005✔
248
249	STypeMod typeGetTypeModFromDataType(const SDataType* pDataType) {	32,820,212✔
250	if (IS_DECIMAL_TYPE(pDataType->type)) return decimalCalcTypeMod(pDataType->precision, pDataType->scale);	32,820,212✔
251	return 0;	32,810,653✔
252	}
253
254	STypeMod typeGetTypeMod(uint8_t type, uint8_t prec, uint8_t scale, int32_t bytes) {	64,468,749✔
255	if (IS_DECIMAL_TYPE(type)) {	64,468,749✔
256	return decimalCalcTypeMod(prec, scale);	12,395,889✔
257	}
258	return 0;	52,072,860✔
259	}
260
261	void fillTypeFromTypeMod(SDataType* pType, STypeMod mod) {	5,025,549✔
262	if (IS_DECIMAL_TYPE(pType->type)) {	5,025,549✔
263	decimalFromTypeMod(mod, &pType->precision, &pType->scale);	20,478✔
264	}
265	}	5,025,565✔
266
267	void extractTypeFromTypeMod(uint8_t type, STypeMod typeMod, uint8_t prec, uint8_t scale, int32_t *bytes) {	5,314,614✔
268	if (IS_DECIMAL_TYPE(type)) {	5,314,614✔
269	decimalFromTypeMod(typeMod, prec, scale);	5,312,915✔
270	} else {
271	if (prec) *prec = 0;	1,699!
272	if (scale) *scale = 0;	1,699!
273	}
274	if (bytes) *bytes = tDataTypes[type].bytes;	5,314,361!
275	}	5,314,361✔
276
UNCOV 277	uint8_t getScaleFromTypeMod(int32_t type, STypeMod mod) {	×
UNCOV 278	if (IS_DECIMAL_TYPE(type)) return (uint8_t)(mod & 0xFF);	×
UNCOV 279	return 0;	×
280	}
281
282	// bytes, 0, prec, scale
283	void fillBytesForDecimalType(int32_t *pBytes, int32_t type, uint8_t precision, uint8_t scale) {	292,304✔
284	*pBytes = 0;	292,304✔
285	*pBytes = tDataTypes[type].bytes << 24;	292,304✔
286	*pBytes \|= (uint32_t)precision << 8;	292,304✔
287	*pBytes \|= scale;	292,304✔
288	}	292,304✔
289
290	void extractDecimalTypeInfoFromBytes(int32_t pBytes, uint8_t precision, uint8_t *scale) {	286,634✔
291	precision = (uint8_t)((pBytes >> 8) & 0xFF);	286,634✔
292	scale = (uint8_t)(pBytes & 0xFF);	286,634✔
293	*pBytes >>= 24;	286,634✔
294	}	286,634✔
295
296	int32_t calcTypeBytesFromSchemaBytes(int32_t type, int32_t schemaBytes, bool isStmt) {	7,660,227✔
297	if (isStmt) return schemaBytes;	7,660,227✔
298	if (type == TSDB_DATA_TYPE_VARCHAR \|\| type == TSDB_DATA_TYPE_VARBINARY \|\| type == TSDB_DATA_TYPE_GEOMETRY) {	7,649,094✔
299	return schemaBytes - VARSTR_HEADER_SIZE;	1,673,411✔
300	} else if (type == TSDB_DATA_TYPE_NCHAR \|\| type == TSDB_DATA_TYPE_JSON) {	5,975,683✔
301	return (schemaBytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;	395,185✔
302	}
303	return schemaBytes;	5,580,498✔
304	}
305
306	int32_t calcSchemaBytesFromTypeBytes(int32_t type, int32_t varTypeBytes, bool isStmt) {	133,146,791✔
307	if (isStmt) return varTypeBytes;	133,146,791✔
308	if (type == TSDB_DATA_TYPE_VARCHAR \|\| type == TSDB_DATA_TYPE_VARBINARY \|\| type == TSDB_DATA_TYPE_GEOMETRY) {	133,145,857✔
309	return varTypeBytes + VARSTR_HEADER_SIZE;	18,260,950✔
310	} else if (type == TSDB_DATA_TYPE_NCHAR \|\| type == TSDB_DATA_TYPE_JSON) {	114,884,907✔
311	return varTypeBytes * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;	8,486,905✔
312	}
313	return varTypeBytes;	106,398,002✔
314	}

taosdata / TDengine / #3660

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous