clientSml.c 54.1 KB
Newer Older
wmmhello's avatar
wmmhello 已提交
1 2 3 4 5 6 7
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "query.h"
#include "taos.h"
wmmhello's avatar
wmmhello 已提交
8 9 10 11
#include "taoserror.h"
#include "tdef.h"
#include "tlog.h"
#include "tmsg.h"
wmmhello's avatar
wmmhello 已提交
12
#include "tstrbuild.h"
wmmhello's avatar
wmmhello 已提交
13 14 15 16
#include "ttime.h"
#include "ttypes.h"
#include "tcommon.h"
#include "catalog.h"
17
#include "clientInt.h"
wmmhello's avatar
wmmhello 已提交
18
//=================================================================================================
wmmhello's avatar
wmmhello 已提交
19 20 21 22 23 24

#define SPACE ' '
#define COMMA ','
#define EQUAL '='
#define QUOTE '"'
#define SLASH '\\'
wmmhello's avatar
wmmhello 已提交
25
#define tsMaxSQLStringLen (1024*1024)
wmmhello's avatar
wmmhello 已提交
26

wmmhello's avatar
wmmhello 已提交
27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116
//=================================================================================================
typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;

typedef enum {
  SCHEMA_ACTION_CREATE_STABLE,
  SCHEMA_ACTION_ADD_COLUMN,
  SCHEMA_ACTION_ADD_TAG,
  SCHEMA_ACTION_CHANGE_COLUMN_SIZE,
  SCHEMA_ACTION_CHANGE_TAG_SIZE,
} ESchemaAction;

typedef struct {
  char sTableName[TSDB_TABLE_NAME_LEN];
  SHashObj *tags;
  SHashObj *fields;
} SCreateSTableActionInfo;

typedef struct {
  char sTableName[TSDB_TABLE_NAME_LEN];
  SSmlKv * field;
} SAlterSTableActionInfo;

typedef struct {
  ESchemaAction action;
  union {
    SCreateSTableActionInfo createSTable;
    SAlterSTableActionInfo alterSTable;
  };
} SSchemaAction;

typedef struct {
  const char* measure;
  const char* tags;
  const char* cols;
  const char* timestamp;

  int32_t measureLen;
  int32_t measureTagsLen;
  int32_t tagsLen;
  int32_t colsLen;
  int32_t timestampLen;
} SSmlLineInfo;

typedef struct {
  const char     *sTableName;   // super table name
  uint8_t        sTableNameLen;
  char           childTableName[TSDB_TABLE_NAME_LEN];
  uint64_t       uid;

  SArray         *tags;

  // colsFormat store cols formated, for quick parse, if info->formatData is true
  SArray         *colsFormat;  // elements are SArray<SSmlKv*>

  // cols & colsColumn store cols un formated
  SArray         *cols;        // elements are SHashObj<cols key string, SSmlKv*> for find by key quickly
  SHashObj       *columnsHash;     // elements are <cols key string, 1>, just for judge if key exists quickly.
} SSmlTableInfo;

typedef struct {
  SHashObj   *tagHash;
  SHashObj   *fieldHash;
  STableMeta *tableMeta;
} SSmlSTableMeta;

typedef struct {
  int32_t   len;
  char      *buf;
} SSmlMsgBuf;

typedef struct {
  uint64_t          id;

  SMLProtocolType   protocol;
  int8_t            precision;
  bool              dataFormat;     // true means that the name, number and order of keys in each line are the same

  SHashObj          *childTables;
  SHashObj          *superTables;
  SHashObj          *pVgHash;
  void              *exec;

  STscObj           *taos;
  SCatalog          *pCatalog;
  SRequestObj       *pRequest;
  SQuery            *pQuery;

  int32_t           affectedRows;
  SSmlMsgBuf        msgBuf;
} SSmlHandle;
wmmhello's avatar
wmmhello 已提交
117 118 119
//=================================================================================================

static uint64_t linesSmlHandleId = 0;
wmmhello's avatar
wmmhello 已提交
120 121
static const char* TS = "_ts";
static const char* TAG = "_tagNone";
wmmhello's avatar
wmmhello 已提交
122

wmmhello's avatar
wmmhello 已提交
123
//=================================================================================================
wmmhello's avatar
wmmhello 已提交
124

wmmhello's avatar
wmmhello 已提交
125
static uint64_t smlGenId() {
wmmhello's avatar
wmmhello 已提交
126 127 128 129 130 131 132 133 134
  uint64_t id;

  do {
    id = atomic_add_fetch_64(&linesSmlHandleId, 1);
  } while (id == 0);

  return id;
}

wmmhello's avatar
wmmhello 已提交
135
static int32_t smlBuildInvalidDataMsg(SSmlMsgBuf* pBuf, const char *msg1, const char *msg2) {
wmmhello's avatar
wmmhello 已提交
136 137 138 139 140 141
  if(msg1) strncat(pBuf->buf, msg1, pBuf->len);
  int32_t left = pBuf->len - strlen(pBuf->buf);
  if(left > 2 && msg2) {
    strncat(pBuf->buf, ":", left - 1);
    strncat(pBuf->buf, msg2, left - 2);
  }
wmmhello's avatar
wmmhello 已提交
142 143 144
  return TSDB_CODE_SML_INVALID_DATA;
}

wmmhello's avatar
wmmhello 已提交
145
static int smlCompareKv(const void* p1, const void* p2) {
146
  SSmlKv* kv1 = (SSmlKv*)p1;
wmmhello's avatar
wmmhello 已提交
147
  SSmlKv* kv2 = (SSmlKv*)p2;
wmmhello's avatar
wmmhello 已提交
148 149
  int32_t kvLen1 = kv1->keyLen;
  int32_t kvLen2 = kv2->keyLen;
wmmhello's avatar
wmmhello 已提交
150
  int32_t res = strncasecmp(kv1->key, kv2->key, TMIN(kvLen1, kvLen2));
wmmhello's avatar
wmmhello 已提交
151 152 153 154 155 156 157
  if (res != 0) {
    return res;
  } else {
    return kvLen1-kvLen2;
  }
}

wmmhello's avatar
wmmhello 已提交
158
static void smlBuildChildTableName(SSmlTableInfo *tags) {
wmmhello's avatar
wmmhello 已提交
159 160
  int32_t size = taosArrayGetSize(tags->tags);
  ASSERT(size > 0);
wmmhello's avatar
wmmhello 已提交
161
  taosArraySort(tags->tags, smlCompareKv);
wmmhello's avatar
wmmhello 已提交
162

wmmhello's avatar
wmmhello 已提交
163 164 165 166 167 168
  SStringBuilder sb = {0};
  taosStringBuilderAppendStringLen(&sb, tags->sTableName, tags->sTableNameLen);
  for (int j = 0; j < size; ++j) {
    SSmlKv *tagKv = taosArrayGetP(tags->tags, j);
    taosStringBuilderAppendStringLen(&sb, tagKv->key, tagKv->keyLen);
    taosStringBuilderAppendStringLen(&sb, tagKv->value, tagKv->valueLen);
wmmhello's avatar
wmmhello 已提交
169 170 171 172 173 174 175 176 177
  }
  size_t len = 0;
  char* keyJoined = taosStringBuilderGetResult(&sb, &len);
  T_MD5_CTX context;
  tMD5Init(&context);
  tMD5Update(&context, (uint8_t *)keyJoined, (uint32_t)len);
  tMD5Final(&context);
  uint64_t digest1 = *(uint64_t*)(context.digest);
  uint64_t digest2 = *(uint64_t*)(context.digest + 8);
wmmhello's avatar
wmmhello 已提交
178
  snprintf(tags->childTableName, TSDB_TABLE_NAME_LEN, "t_%016"PRIx64"%016"PRIx64, digest1, digest2);
wmmhello's avatar
wmmhello 已提交
179
  taosStringBuilderDestroy(&sb);
wmmhello's avatar
wmmhello 已提交
180
  tags->uid = digest1;
wmmhello's avatar
wmmhello 已提交
181 182
}

wmmhello's avatar
wmmhello 已提交
183 184
static int32_t smlGenerateSchemaAction(SSchema* pointColField, SHashObj* dbAttrHash, SArray* dbAttrArray, bool isTag, char sTableName[],
                                       SSchemaAction* action, bool* actionNeeded, SSmlHandle* info) {
wmmhello's avatar
wmmhello 已提交
185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223
//  char fieldName[TSDB_COL_NAME_LEN] = {0};
//  strcpy(fieldName, pointColField->name);
//
//  size_t* pDbIndex = taosHashGet(dbAttrHash, fieldName, strlen(fieldName));
//  if (pDbIndex) {
//    SSchema* dbAttr = taosArrayGet(dbAttrArray, *pDbIndex);
//    assert(strcasecmp(dbAttr->name, pointColField->name) == 0);
//    if (pointColField->type != dbAttr->type) {
//      uError("SML:0x%"PRIx64" point type and db type mismatch. key: %s. point type: %d, db type: %d", info->id, pointColField->name,
//               pointColField->type, dbAttr->type);
//      return TSDB_CODE_TSC_INVALID_VALUE;
//    }
//
//    if (IS_VAR_DATA_TYPE(pointColField->type) && (pointColField->bytes > dbAttr->bytes)) {
//      if (isTag) {
//        action->action = SCHEMA_ACTION_CHANGE_TAG_SIZE;
//      } else {
//        action->action = SCHEMA_ACTION_CHANGE_COLUMN_SIZE;
//      }
//      memset(&action->alterSTable, 0,  sizeof(SAlterSTableActionInfo));
//      memcpy(action->alterSTable.sTableName, sTableName, TSDB_TABLE_NAME_LEN);
//      action->alterSTable.field = pointColField;
//      *actionNeeded = true;
//    }
//  } else {
//    if (isTag) {
//      action->action = SCHEMA_ACTION_ADD_TAG;
//    } else {
//      action->action = SCHEMA_ACTION_ADD_COLUMN;
//    }
//    memset(&action->alterSTable, 0, sizeof(SAlterSTableActionInfo));
//    memcpy(action->alterSTable.sTableName, sTableName, TSDB_TABLE_NAME_LEN);
//    action->alterSTable.field = pointColField;
//    *actionNeeded = true;
//  }
//  if (*actionNeeded) {
//    uDebug("SML:0x%" PRIx64 " generate schema action. column name: %s, action: %d", info->id, fieldName,
//             action->action);
//  }
wmmhello's avatar
wmmhello 已提交
224 225 226
  return 0;
}

wmmhello's avatar
wmmhello 已提交
227
static int32_t smlBuildColumnDescription(SSmlKv* field, char* buf, int32_t bufSize, int32_t* outBytes) {
wmmhello's avatar
wmmhello 已提交
228
  uint8_t type = field->type;
wmmhello's avatar
wmmhello 已提交
229 230
  char    tname[TSDB_TABLE_NAME_LEN] = {0};
  memcpy(tname, field->key, field->keyLen);
wmmhello's avatar
wmmhello 已提交
231
  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
232
    int32_t bytes = field->valueLen;   // todo
wmmhello's avatar
wmmhello 已提交
233
    int out = snprintf(buf, bufSize,"%s %s(%d)",
wmmhello's avatar
wmmhello 已提交
234
                       tname,tDataTypes[field->type].name, bytes);
wmmhello's avatar
wmmhello 已提交
235 236
    *outBytes = out;
  } else {
wmmhello's avatar
wmmhello 已提交
237
    int out = snprintf(buf, bufSize, "%s %s", tname, tDataTypes[type].name);
wmmhello's avatar
wmmhello 已提交
238 239 240 241 242 243
    *outBytes = out;
  }

  return 0;
}

wmmhello's avatar
wmmhello 已提交
244
static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) {
wmmhello's avatar
wmmhello 已提交
245 246
  int32_t code = 0;
  int32_t outBytes = 0;
wmmhello's avatar
wmmhello 已提交
247
  char *result = (char *)taosMemoryCalloc(1, tsMaxSQLStringLen+1);
wmmhello's avatar
wmmhello 已提交
248 249 250 251 252 253
  int32_t capacity = tsMaxSQLStringLen +  1;

  uDebug("SML:0x%"PRIx64" apply schema action. action: %d", info->id, action->action);
  switch (action->action) {
    case SCHEMA_ACTION_ADD_COLUMN: {
      int n = sprintf(result, "alter stable %s add column ", action->alterSTable.sTableName);
wmmhello's avatar
wmmhello 已提交
254 255
      smlBuildColumnDescription(action->alterSTable.field, result+n, capacity-n, &outBytes);
      TAOS_RES* res = taos_query(info->taos, result); //TODO async doAsyncQuery
wmmhello's avatar
wmmhello 已提交
256
      code = taos_errno(res);
wmmhello's avatar
wmmhello 已提交
257
      const char* errStr = taos_errstr(res);
wmmhello's avatar
wmmhello 已提交
258 259 260 261 262 263 264
      char* begin = strstr(errStr, "duplicated column names");
      bool tscDupColNames = (begin != NULL);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error: %s", info->id, errStr);
      }
      taos_free_result(res);

wmmhello's avatar
wmmhello 已提交
265 266
//      if (code == TSDB_CODE_MND_FIELD_ALREADY_EXIST || code == TSDB_CODE_MND_TAG_ALREADY_EXIST || tscDupColNames) {
      if (code == TSDB_CODE_MND_TAG_ALREADY_EXIST || tscDupColNames) {
wmmhello's avatar
wmmhello 已提交
267
        TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE");
wmmhello's avatar
wmmhello 已提交
268 269 270 271 272 273 274 275 276 277 278
        code = taos_errno(res2);
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
        }
        taos_free_result(res2);
        taosMsleep(500);
      }
      break;
    }
    case SCHEMA_ACTION_ADD_TAG: {
      int n = sprintf(result, "alter stable %s add tag ", action->alterSTable.sTableName);
wmmhello's avatar
wmmhello 已提交
279
      smlBuildColumnDescription(action->alterSTable.field,
wmmhello's avatar
wmmhello 已提交
280
                             result+n, capacity-n, &outBytes);
wmmhello's avatar
wmmhello 已提交
281
      TAOS_RES* res = taos_query(info->taos, result); //TODO async doAsyncQuery
wmmhello's avatar
wmmhello 已提交
282
      code = taos_errno(res);
wmmhello's avatar
wmmhello 已提交
283
      const char* errStr = taos_errstr(res);
wmmhello's avatar
wmmhello 已提交
284 285 286 287 288 289 290
      char* begin = strstr(errStr, "duplicated column names");
      bool tscDupColNames = (begin != NULL);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res));
      }
      taos_free_result(res);

wmmhello's avatar
wmmhello 已提交
291 292
//      if (code ==TSDB_CODE_MND_TAG_ALREADY_EXIST || code == TSDB_CODE_MND_FIELD_ALREAY_EXIST || tscDupColNames) {
      if (code ==TSDB_CODE_MND_TAG_ALREADY_EXIST || tscDupColNames) {
wmmhello's avatar
wmmhello 已提交
293
        TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE");
wmmhello's avatar
wmmhello 已提交
294 295 296 297 298 299 300 301 302 303 304
        code = taos_errno(res2);
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
        }
        taos_free_result(res2);
        taosMsleep(500);
      }
      break;
    }
    case SCHEMA_ACTION_CHANGE_COLUMN_SIZE: {
      int n = sprintf(result, "alter stable %s modify column ", action->alterSTable.sTableName);
wmmhello's avatar
wmmhello 已提交
305
      smlBuildColumnDescription(action->alterSTable.field, result+n,
wmmhello's avatar
wmmhello 已提交
306
                             capacity-n, &outBytes);
wmmhello's avatar
wmmhello 已提交
307
      TAOS_RES* res = taos_query(info->taos, result); //TODO async doAsyncQuery
wmmhello's avatar
wmmhello 已提交
308 309 310 311 312 313
      code = taos_errno(res);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res));
      }
      taos_free_result(res);

wmmhello's avatar
wmmhello 已提交
314 315
//      if (code == TSDB_CODE_MND_INVALID_COLUMN_LENGTH || code == TSDB_CODE_TSC_INVALID_COLUMN_LENGTH) {
      if (code == TSDB_CODE_TSC_INVALID_COLUMN_LENGTH) {
wmmhello's avatar
wmmhello 已提交
316
        TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE");
wmmhello's avatar
wmmhello 已提交
317 318 319 320 321 322 323 324 325 326 327
        code = taos_errno(res2);
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
        }
        taos_free_result(res2);
        taosMsleep(500);
      }
      break;
    }
    case SCHEMA_ACTION_CHANGE_TAG_SIZE: {
      int n = sprintf(result, "alter stable %s modify tag ", action->alterSTable.sTableName);
wmmhello's avatar
wmmhello 已提交
328
      smlBuildColumnDescription(action->alterSTable.field, result+n,
wmmhello's avatar
wmmhello 已提交
329
                             capacity-n, &outBytes);
wmmhello's avatar
wmmhello 已提交
330
      TAOS_RES* res = taos_query(info->taos, result); //TODO async doAsyncQuery
wmmhello's avatar
wmmhello 已提交
331 332 333 334 335 336
      code = taos_errno(res);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res));
      }
      taos_free_result(res);

wmmhello's avatar
wmmhello 已提交
337 338
//      if (code == TSDB_CODE_MND_INVALID_TAG_LENGTH || code == TSDB_CODE_TSC_INVALID_TAG_LENGTH) {
      if (code == TSDB_CODE_TSC_INVALID_TAG_LENGTH) {
wmmhello's avatar
wmmhello 已提交
339
        TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE");
wmmhello's avatar
wmmhello 已提交
340 341 342 343 344 345 346 347 348 349 350 351
        code = taos_errno(res2);
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
        }
        taos_free_result(res2);
        taosMsleep(500);
      }
      break;
    }
    case SCHEMA_ACTION_CREATE_STABLE: {
      int n = sprintf(result, "create stable %s (", action->createSTable.sTableName);
      char* pos = result + n; int freeBytes = capacity - n;
wmmhello's avatar
wmmhello 已提交
352

wmmhello's avatar
wmmhello 已提交
353 354
      size_t size = taosHashGetSize(action->createSTable.fields);
      SArray *cols = taosArrayInit(size, POINTER_BYTES);
wmmhello's avatar
wmmhello 已提交
355
      SSmlKv **kv = taosHashIterate(action->createSTable.fields, NULL);
wmmhello's avatar
wmmhello 已提交
356
      while(kv){
wmmhello's avatar
wmmhello 已提交
357 358 359 360 361 362 363 364 365 366 367
        if(strncmp((*kv)->key, TS, strlen(TS)) == 0 && (*kv)->type == TSDB_DATA_TYPE_TIMESTAMP){
          taosArrayInsert(cols, 0, kv);
        }else{
          taosArrayPush(cols, kv);
        }
        kv = taosHashIterate(action->createSTable.fields, kv);
      }

      for(int i = 0; i < taosArrayGetSize(cols); i++){
        SSmlKv *kvNew = taosArrayGetP(cols, i);
        smlBuildColumnDescription(kvNew, pos, freeBytes, &outBytes);
wmmhello's avatar
wmmhello 已提交
368 369 370
        pos += outBytes; freeBytes -= outBytes;
        *pos = ','; ++pos; --freeBytes;
      }
wmmhello's avatar
wmmhello 已提交
371 372
      taosArrayDestroy(cols);

wmmhello's avatar
wmmhello 已提交
373 374 375 376 377
      --pos; ++freeBytes;

      outBytes = snprintf(pos, freeBytes, ") tags (");
      pos += outBytes; freeBytes -= outBytes;

wmmhello's avatar
wmmhello 已提交
378 379
      kv = taosHashIterate(action->createSTable.tags, NULL);
      while(kv){
wmmhello's avatar
wmmhello 已提交
380
        smlBuildColumnDescription(*kv, pos, freeBytes, &outBytes);
wmmhello's avatar
wmmhello 已提交
381 382
        pos += outBytes; freeBytes -= outBytes;
        *pos = ','; ++pos; --freeBytes;
wmmhello's avatar
wmmhello 已提交
383
        kv = taosHashIterate(action->createSTable.tags, kv);
wmmhello's avatar
wmmhello 已提交
384 385 386
      }
      pos--; ++freeBytes;
      outBytes = snprintf(pos, freeBytes, ")");
wmmhello's avatar
wmmhello 已提交
387
      TAOS_RES* res = taos_query(info->taos, result);
wmmhello's avatar
wmmhello 已提交
388 389 390 391 392 393
      code = taos_errno(res);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res));
      }
      taos_free_result(res);

wmmhello's avatar
wmmhello 已提交
394
      if (code == TSDB_CODE_MND_STB_ALREADY_EXIST) {
wmmhello's avatar
wmmhello 已提交
395
        TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE");
wmmhello's avatar
wmmhello 已提交
396 397 398 399 400 401 402 403 404 405 406 407 408 409
        code = taos_errno(res2);
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2));
        }
        taos_free_result(res2);
        taosMsleep(500);
      }
      break;
    }

    default:
      break;
  }

wmmhello's avatar
wmmhello 已提交
410
  taosMemoryFreeClear(result);
wmmhello's avatar
wmmhello 已提交
411 412 413 414 415 416
  if (code != 0) {
    uError("SML:0x%"PRIx64 " apply schema action failure. %s", info->id, tstrerror(code));
  }
  return code;
}

wmmhello's avatar
wmmhello 已提交
417
static int32_t smlModifyDBSchemas(SSmlHandle* info) {
wmmhello's avatar
wmmhello 已提交
418 419 420 421 422 423 424
  int32_t code = 0;

  SSmlSTableMeta** tableMetaSml = taosHashIterate(info->superTables, NULL);
  while (tableMetaSml) {
    SSmlSTableMeta* cTablePoints = *tableMetaSml;

    STableMeta *pTableMeta = NULL;
wmmhello's avatar
wmmhello 已提交
425
    SEpSet ep = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
wmmhello's avatar
wmmhello 已提交
426

wmmhello's avatar
wmmhello 已提交
427
    size_t superTableLen = 0;
wmmhello's avatar
wmmhello 已提交
428
    void *superTable = taosHashGetKey(tableMetaSml, &superTableLen);    // todo escape
wmmhello's avatar
wmmhello 已提交
429 430 431
    SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
    strcpy(pName.dbname, info->pRequest->pDb);
    memcpy(pName.tname, superTable, superTableLen);
wmmhello's avatar
wmmhello 已提交
432

wmmhello's avatar
wmmhello 已提交
433
    code = catalogGetSTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, &pTableMeta);
wmmhello's avatar
wmmhello 已提交
434

wmmhello's avatar
wmmhello 已提交
435
    if (code == TSDB_CODE_TDB_INVALID_TABLE_ID || code == TSDB_CODE_MND_INVALID_STB) {
wmmhello's avatar
wmmhello 已提交
436 437
      SSchemaAction schemaAction = {0};
      schemaAction.action = SCHEMA_ACTION_CREATE_STABLE;
wmmhello's avatar
wmmhello 已提交
438 439 440
      memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);
      schemaAction.createSTable.tags = cTablePoints->tagHash;
      schemaAction.createSTable.fields = cTablePoints->fieldHash;
wmmhello's avatar
wmmhello 已提交
441 442 443 444 445 446
      code = smlApplySchemaAction(info, &schemaAction);
      if (code != 0) {
        uError("SML:0x%"PRIx64" smlApplySchemaAction failed. can not create %s", info->id, schemaAction.createSTable.sTableName);
        return code;
      }

wmmhello's avatar
wmmhello 已提交
447
      code = catalogGetSTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, &pTableMeta);
wmmhello's avatar
wmmhello 已提交
448
      if (code != 0) {
wmmhello's avatar
wmmhello 已提交
449
        uError("SML:0x%"PRIx64" catalogGetSTableMeta failed. super table name %s", info->id, schemaAction.createSTable.sTableName);
wmmhello's avatar
wmmhello 已提交
450 451
        return code;
      }
wmmhello's avatar
wmmhello 已提交
452
    }else if (code == TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
453 454 455 456
    } else {
      uError("SML:0x%"PRIx64" load table meta error: %s", info->id, tstrerror(code));
      return code;
    }
wmmhello's avatar
wmmhello 已提交
457
    cTablePoints->tableMeta = pTableMeta;
wmmhello's avatar
wmmhello 已提交
458

wmmhello's avatar
wmmhello 已提交
459
    tableMetaSml = taosHashIterate(info->superTables, tableMetaSml);
wmmhello's avatar
wmmhello 已提交
460 461 462 463 464 465 466 467 468 469 470
  }
  return 0;
}

//=========================================================================

/*        Field                          Escape charaters
    1: measurement                        Comma,Space
    2: tag_key, tag_value, field_key  Comma,Equal Sign,Space
    3: field_value                    Double quote,Backslash
*/
wmmhello's avatar
wmmhello 已提交
471 472 473 474 475 476 477 478 479 480 481 482 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
//static void escapeSpecialCharacter(uint8_t field, const char **pos) {
//  const char *cur = *pos;
//  if (*cur != '\\') {
//    return;
//  }
//  switch (field) {
//    case 1:
//      switch (*(cur + 1)) {
//        case ',':
//        case ' ':
//          cur++;
//          break;
//        default:
//          break;
//      }
//      break;
//    case 2:
//      switch (*(cur + 1)) {
//        case ',':
//        case ' ':
//        case '=':
//          cur++;
//          break;
//        default:
//          break;
//      }
//      break;
//    case 3:
//      switch (*(cur + 1)) {
//        case '"':
//        case '\\':
//          cur++;
//          break;
//        default:
//          break;
//      }
//      break;
//    default:
//      break;
//  }
//  *pos = cur;
//}
wmmhello's avatar
wmmhello 已提交
513

wmmhello's avatar
wmmhello 已提交
514
static bool smlParseTinyInt(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
515 516
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
wmmhello's avatar
wmmhello 已提交
517 518 519
  if (len <= 2) {
    return false;
  }
wmmhello's avatar
wmmhello 已提交
520
  const char *signalPos = pVal + len - 2;
wmmhello's avatar
wmmhello 已提交
521
  if (!strncasecmp(signalPos, "i8", 2)) {
wmmhello's avatar
wmmhello 已提交
522 523 524 525
    char *endptr = NULL;
    int64_t result = strtoll(pVal, &endptr, 10);
    if(endptr != signalPos){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
526
      smlBuildInvalidDataMsg(msg, "invalid tiny int", endptr);
wmmhello's avatar
wmmhello 已提交
527 528
    }else if(!IS_VALID_TINYINT(result)){
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
529
      smlBuildInvalidDataMsg(msg, "tiny int out of range[-128,127]", endptr);
wmmhello's avatar
wmmhello 已提交
530 531 532 533
    }else{
      kvVal->i = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
534 535 536 537 538
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
539
static bool smlParseTinyUint(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
540 541
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
wmmhello's avatar
wmmhello 已提交
542 543 544 545 546 547
  if (len <= 2) {
    return false;
  }
  if (pVal[0] == '-') {
    return false;
  }
wmmhello's avatar
wmmhello 已提交
548
  const char *signalPos = pVal + len - 2;
wmmhello's avatar
wmmhello 已提交
549
  if (!strncasecmp(signalPos, "u8", 2)) {
wmmhello's avatar
wmmhello 已提交
550 551 552 553
    char *endptr = NULL;
    int64_t result = strtoll(pVal, &endptr, 10);
    if(endptr != signalPos){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
554
      smlBuildInvalidDataMsg(msg, "invalid unsigned tiny int", endptr);
wmmhello's avatar
wmmhello 已提交
555 556
    }else if(!IS_VALID_UTINYINT(result)){
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
557
      smlBuildInvalidDataMsg(msg, "unsigned tiny int out of range[0,255]", endptr);
wmmhello's avatar
wmmhello 已提交
558 559 560 561
    }else{
      kvVal->i = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
562 563 564 565 566
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
567
static bool smlParseSmallInt(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
568 569
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
wmmhello's avatar
wmmhello 已提交
570 571 572
  if (len <= 3) {
    return false;
  }
wmmhello's avatar
wmmhello 已提交
573
  const char *signalPos = pVal + len - 3;
wmmhello's avatar
wmmhello 已提交
574
  if (!strncasecmp(signalPos, "i16", 3)) {
wmmhello's avatar
wmmhello 已提交
575 576 577 578
    char *endptr = NULL;
    int64_t result = strtoll(pVal, &endptr, 10);
    if(endptr != signalPos){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
579
      smlBuildInvalidDataMsg(msg, "invalid small int", endptr);
wmmhello's avatar
wmmhello 已提交
580 581
    }else if(!IS_VALID_SMALLINT(result)){
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
582
      smlBuildInvalidDataMsg(msg, "small int our of range[-32768,32767]", endptr);
wmmhello's avatar
wmmhello 已提交
583 584 585 586
    }else{
      kvVal->i = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
587 588 589 590 591
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
592
static bool smlParseSmallUint(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
593 594
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
wmmhello's avatar
wmmhello 已提交
595 596 597 598 599 600
  if (len <= 3) {
    return false;
  }
  if (pVal[0] == '-') {
    return false;
  }
wmmhello's avatar
wmmhello 已提交
601
  const char *signalPos = pVal + len - 3;
wmmhello's avatar
wmmhello 已提交
602
  if (strncasecmp(signalPos, "u16", 3) == 0) {
wmmhello's avatar
wmmhello 已提交
603 604 605 606
    char *endptr = NULL;
    int64_t result = strtoll(pVal, &endptr, 10);
    if(endptr != signalPos){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
607
      smlBuildInvalidDataMsg(msg, "invalid unsigned small int", endptr);
wmmhello's avatar
wmmhello 已提交
608 609
    }else if(!IS_VALID_USMALLINT(result)){
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
610
      smlBuildInvalidDataMsg(msg, "unsigned small int out of rang[0,65535]", endptr);
wmmhello's avatar
wmmhello 已提交
611 612 613 614
    }else{
      kvVal->i = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
615 616 617 618 619
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
620
static bool smlParseInt(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
621 622
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
wmmhello's avatar
wmmhello 已提交
623 624 625
  if (len <= 3) {
    return false;
  }
wmmhello's avatar
wmmhello 已提交
626
  const char *signalPos = pVal + len - 3;
wmmhello's avatar
wmmhello 已提交
627
  if (strncasecmp(signalPos, "i32", 3) == 0) {
wmmhello's avatar
wmmhello 已提交
628 629 630 631
    char *endptr = NULL;
    int64_t result = strtoll(pVal, &endptr, 10);
    if(endptr != signalPos){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
632
      smlBuildInvalidDataMsg(msg, "invalid int", endptr);
wmmhello's avatar
wmmhello 已提交
633 634
    }else if(!IS_VALID_INT(result)){
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
635
      smlBuildInvalidDataMsg(msg, "int out of range[-2147483648,2147483647]", endptr);
wmmhello's avatar
wmmhello 已提交
636 637 638 639
    }else{
      kvVal->i = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
640 641 642 643 644
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
645
static bool smlParseUint(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
646 647
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
wmmhello's avatar
wmmhello 已提交
648 649 650 651 652 653
  if (len <= 3) {
    return false;
  }
  if (pVal[0] == '-') {
    return false;
  }
wmmhello's avatar
wmmhello 已提交
654
  const char *signalPos = pVal + len - 3;
wmmhello's avatar
wmmhello 已提交
655
  if (strncasecmp(signalPos, "u32", 3) == 0) {
wmmhello's avatar
wmmhello 已提交
656 657 658 659
    char *endptr = NULL;
    int64_t result = strtoll(pVal, &endptr, 10);
    if(endptr != signalPos){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
660
      smlBuildInvalidDataMsg(msg, "invalid unsigned int", endptr);
wmmhello's avatar
wmmhello 已提交
661 662
    }else if(!IS_VALID_UINT(result)){
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
663
      smlBuildInvalidDataMsg(msg, "unsigned int out of range[0,4294967295]", endptr);
wmmhello's avatar
wmmhello 已提交
664 665 666 667
    }else{
      kvVal->i = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
668 669 670 671 672
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
673
static bool smlParseBigInt(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
674 675
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
wmmhello's avatar
wmmhello 已提交
676
  if (len > 3 && strncasecmp(pVal + len - 3, "i64", 3) == 0) {
wmmhello's avatar
wmmhello 已提交
677
    char *endptr = NULL;
wmmhello's avatar
wmmhello 已提交
678
    errno = 0;
wmmhello's avatar
wmmhello 已提交
679 680 681
    int64_t result = strtoll(pVal, &endptr, 10);
    if(endptr != pVal + len - 3){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
682
      smlBuildInvalidDataMsg(msg, "invalid big int", endptr);
wmmhello's avatar
wmmhello 已提交
683
    }else if(errno == ERANGE || !IS_VALID_BIGINT(result)){
wmmhello's avatar
wmmhello 已提交
684
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
685
      smlBuildInvalidDataMsg(msg, "big int out of range[-9223372036854775808,9223372036854775807]", endptr);
wmmhello's avatar
wmmhello 已提交
686 687 688 689 690 691 692
    }else{
      kvVal->i = result;
      *isValid = true;
    }
    return true;
  }else if (len > 1 && pVal[len - 1] == 'i') {
    char *endptr = NULL;
wmmhello's avatar
wmmhello 已提交
693
    errno = 0;
wmmhello's avatar
wmmhello 已提交
694 695 696
    int64_t result = strtoll(pVal, &endptr, 10);
    if(endptr != pVal + len - 1){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
697
      smlBuildInvalidDataMsg(msg, "invalid big int", endptr);
wmmhello's avatar
wmmhello 已提交
698
    }else if(errno == ERANGE || !IS_VALID_BIGINT(result)){
wmmhello's avatar
wmmhello 已提交
699
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
700
      smlBuildInvalidDataMsg(msg, "big int out of range[-9223372036854775808,9223372036854775807]", endptr);
wmmhello's avatar
wmmhello 已提交
701 702 703 704
    }else{
      kvVal->i = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
705 706 707 708 709
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
710
static bool smlParseBigUint(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
711 712
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
wmmhello's avatar
wmmhello 已提交
713 714 715 716 717 718
  if (len <= 3) {
    return false;
  }
  if (pVal[0] == '-') {
    return false;
  }
wmmhello's avatar
wmmhello 已提交
719
  const char *signalPos = pVal + len - 3;
wmmhello's avatar
wmmhello 已提交
720
  if (strncasecmp(signalPos, "u64", 3) == 0) {
wmmhello's avatar
wmmhello 已提交
721
    char *endptr = NULL;
wmmhello's avatar
wmmhello 已提交
722
    errno = 0;
wmmhello's avatar
wmmhello 已提交
723 724 725
    uint64_t result = strtoull(pVal, &endptr, 10);
    if(endptr != signalPos){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
726
      smlBuildInvalidDataMsg(msg, "invalid unsigned big int", endptr);
wmmhello's avatar
wmmhello 已提交
727
    }else if(errno == ERANGE || !IS_VALID_UBIGINT(result)){
wmmhello's avatar
wmmhello 已提交
728
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
729
      smlBuildInvalidDataMsg(msg, "unsigned big int out of range[0,18446744073709551615]", endptr);
wmmhello's avatar
wmmhello 已提交
730 731 732 733
    }else{
      kvVal->u = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
734 735 736 737 738
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
739
static bool smlParseFloat(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
740 741 742
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
  char *endptr = NULL;
wmmhello's avatar
wmmhello 已提交
743
  errno = 0;
wmmhello's avatar
wmmhello 已提交
744
  float result = strtof(pVal, &endptr);
wmmhello's avatar
wmmhello 已提交
745
  if(endptr == pVal + len && errno != ERANGE && IS_VALID_FLOAT(result)){       // 78
wmmhello's avatar
wmmhello 已提交
746 747 748
    kvVal->f = result;
    *isValid = true;
    return true;
wmmhello's avatar
wmmhello 已提交
749
  }
wmmhello's avatar
wmmhello 已提交
750

wmmhello's avatar
wmmhello 已提交
751
  if (len > 3 && strncasecmp(pVal + len - 3, "f32", 3) == 0) {
wmmhello's avatar
wmmhello 已提交
752 753
    if(endptr != pVal + len - 3){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
754
      smlBuildInvalidDataMsg(msg, "invalid float", endptr);
wmmhello's avatar
wmmhello 已提交
755
    }else if(errno == ERANGE || !IS_VALID_FLOAT(result)){
wmmhello's avatar
wmmhello 已提交
756
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
757
      smlBuildInvalidDataMsg(msg, "float out of range[-3.402823466e+38,3.402823466e+38]", endptr);
wmmhello's avatar
wmmhello 已提交
758 759 760 761
    }else{
      kvVal->f = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
762 763 764 765 766
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
767
static bool smlParseDouble(SSmlKv *kvVal, bool *isValid, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
768 769
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
wmmhello's avatar
wmmhello 已提交
770 771 772
  if (len <= 3) {
    return false;
  }
wmmhello's avatar
wmmhello 已提交
773
  const char *signalPos = pVal + len - 3;
wmmhello's avatar
wmmhello 已提交
774
  if (strncasecmp(signalPos, "f64", 3) == 0) {
wmmhello's avatar
wmmhello 已提交
775
    char *endptr = NULL;
wmmhello's avatar
wmmhello 已提交
776
    errno = 0;
wmmhello's avatar
wmmhello 已提交
777 778 779
    double result = strtod(pVal, &endptr);
    if(endptr != signalPos){       // 78ri8
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
780
      smlBuildInvalidDataMsg(msg, "invalid double", endptr);
wmmhello's avatar
wmmhello 已提交
781
    }else if(errno == ERANGE || !IS_VALID_DOUBLE(result)){
wmmhello's avatar
wmmhello 已提交
782
      *isValid = false;
wmmhello's avatar
wmmhello 已提交
783
      smlBuildInvalidDataMsg(msg, "double out of range[-1.7976931348623158e+308,1.7976931348623158e+308]", endptr);
wmmhello's avatar
wmmhello 已提交
784 785 786 787
    }else{
      kvVal->d = result;
      *isValid = true;
    }
wmmhello's avatar
wmmhello 已提交
788 789 790 791 792
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
793
static bool smlParseBool(SSmlKv *kvVal) {
wmmhello's avatar
wmmhello 已提交
794 795 796 797
  const char *pVal = kvVal->value;
  int32_t len = kvVal->valueLen;
  if ((len == 1) && pVal[len - 1] == 't') {
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
798 799 800
    return true;
  }

wmmhello's avatar
wmmhello 已提交
801 802
  if ((len == 1) && pVal[len - 1] == 'f') {
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
803 804 805
    return true;
  }

wmmhello's avatar
wmmhello 已提交
806
  if((len == 4) && !strncasecmp(pVal, "true", len)) {
wmmhello's avatar
wmmhello 已提交
807
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
808 809
    return true;
  }
wmmhello's avatar
wmmhello 已提交
810
  if((len == 5) && !strncasecmp(pVal, "false", len)) {
wmmhello's avatar
wmmhello 已提交
811
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
812 813 814 815 816
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
817
static bool smlIsBinary(const char *pVal, uint16_t len) {
wmmhello's avatar
wmmhello 已提交
818 819 820 821 822 823 824 825 826 827
  //binary: "abc"
  if (len < 2) {
    return false;
  }
  if (pVal[0] == '"' && pVal[len - 1] == '"') {
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
828
static bool smlIsNchar(const char *pVal, uint16_t len) {
wmmhello's avatar
wmmhello 已提交
829 830 831 832 833 834 835 836 837 838
  //nchar: L"abc"
  if (len < 3) {
    return false;
  }
  if ((pVal[0] == 'l' || pVal[0] == 'L')&& pVal[1] == '"' && pVal[len - 1] == '"') {
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
839
static bool smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
840 841
  // put high probability matching type first
  bool isValid = false;
wmmhello's avatar
wmmhello 已提交
842

wmmhello's avatar
wmmhello 已提交
843
  //binary
wmmhello's avatar
wmmhello 已提交
844
  if (smlIsBinary(pVal->value, pVal->valueLen)) {
wmmhello's avatar
wmmhello 已提交
845 846
    pVal->type = TSDB_DATA_TYPE_BINARY;
    pVal->valueLen -= 2;
wmmhello's avatar
wmmhello 已提交
847 848
    pVal->length = pVal->valueLen;
    pVal->value++;
wmmhello's avatar
wmmhello 已提交
849
    return true;
wmmhello's avatar
wmmhello 已提交
850
  }
wmmhello's avatar
wmmhello 已提交
851
  //nchar
wmmhello's avatar
wmmhello 已提交
852
  if (smlIsNchar(pVal->value, pVal->valueLen)) {
wmmhello's avatar
wmmhello 已提交
853
    pVal->type = TSDB_DATA_TYPE_NCHAR;
wmmhello's avatar
wmmhello 已提交
854 855 856 857 858 859 860 861 862 863
    pVal->valueLen -= 3;
    pVal->length = pVal->valueLen;
    pVal->value += 2;
    return true;
  }
  //float
  if (smlParseFloat(pVal, &isValid, msg)) {
    if(!isValid) return false;
    pVal->type = TSDB_DATA_TYPE_FLOAT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
864 865
    return true;
  }
wmmhello's avatar
wmmhello 已提交
866
  //double
wmmhello's avatar
wmmhello 已提交
867
  if (smlParseDouble(pVal, &isValid, msg)) {
wmmhello's avatar
wmmhello 已提交
868 869 870
    if(!isValid) return false;
    pVal->type = TSDB_DATA_TYPE_DOUBLE;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
871

wmmhello's avatar
wmmhello 已提交
872 873 874
    return true;
  }
  //bool
wmmhello's avatar
wmmhello 已提交
875
  if (smlParseBool(pVal)) {
wmmhello's avatar
wmmhello 已提交
876
    pVal->type = TSDB_DATA_TYPE_BOOL;
wmmhello's avatar
wmmhello 已提交
877 878 879
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }
wmmhello's avatar
wmmhello 已提交
880

wmmhello's avatar
wmmhello 已提交
881
  if (smlParseTinyInt(pVal, &isValid, msg)) {
wmmhello's avatar
wmmhello 已提交
882
    if(!isValid) return false;
wmmhello's avatar
wmmhello 已提交
883 884 885 886
    pVal->type = TSDB_DATA_TYPE_TINYINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }
wmmhello's avatar
wmmhello 已提交
887
  if (smlParseTinyUint(pVal, &isValid, msg)) {
wmmhello's avatar
wmmhello 已提交
888
    if(!isValid) return false;
wmmhello's avatar
wmmhello 已提交
889 890 891 892
    pVal->type = TSDB_DATA_TYPE_UTINYINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }
wmmhello's avatar
wmmhello 已提交
893
  if (smlParseSmallInt(pVal, &isValid, msg)) {
wmmhello's avatar
wmmhello 已提交
894
    if(!isValid) return false;
wmmhello's avatar
wmmhello 已提交
895 896 897 898
    pVal->type = TSDB_DATA_TYPE_SMALLINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }
wmmhello's avatar
wmmhello 已提交
899
  if (smlParseSmallUint(pVal, &isValid, msg)) {
wmmhello's avatar
wmmhello 已提交
900
    if(!isValid) return false;
wmmhello's avatar
wmmhello 已提交
901 902 903 904
    pVal->type = TSDB_DATA_TYPE_USMALLINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }
wmmhello's avatar
wmmhello 已提交
905
  if (smlParseInt(pVal, &isValid, msg)) {
wmmhello's avatar
wmmhello 已提交
906
    if(!isValid) return false;
wmmhello's avatar
wmmhello 已提交
907 908 909 910
    pVal->type = TSDB_DATA_TYPE_INT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }
wmmhello's avatar
wmmhello 已提交
911
  if (smlParseUint(pVal, &isValid, msg)) {
wmmhello's avatar
wmmhello 已提交
912
    if(!isValid) return false;
wmmhello's avatar
wmmhello 已提交
913 914 915 916
    pVal->type = TSDB_DATA_TYPE_UINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }
wmmhello's avatar
wmmhello 已提交
917
  if (smlParseBigInt(pVal, &isValid, msg)) {
wmmhello's avatar
wmmhello 已提交
918
    if(!isValid) return false;
wmmhello's avatar
wmmhello 已提交
919 920 921 922
    pVal->type = TSDB_DATA_TYPE_BIGINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }
wmmhello's avatar
wmmhello 已提交
923
  if (smlParseBigUint(pVal, &isValid, msg)) {
wmmhello's avatar
wmmhello 已提交
924
    if(!isValid) return false;
wmmhello's avatar
wmmhello 已提交
925 926 927 928 929
    pVal->type = TSDB_DATA_TYPE_UBIGINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }

wmmhello's avatar
wmmhello 已提交
930
  smlBuildInvalidDataMsg(msg, "invalid data", pVal->value);
wmmhello's avatar
wmmhello 已提交
931 932 933
  return false;
}

wmmhello's avatar
wmmhello 已提交
934
static bool checkDuplicateKey(char *key, SHashObj *pHash, SSmlHandle* info) {
wmmhello's avatar
wmmhello 已提交
935 936 937 938 939
  char *val = NULL;
  val = taosHashGet(pHash, key, strlen(key));
  if (val) {
    uError("SML:0x%"PRIx64" Duplicate key detected:%s", info->id, key);
    return true;
wmmhello's avatar
wmmhello 已提交
940 941
  }

wmmhello's avatar
wmmhello 已提交
942 943
  uint8_t dummy_val = 0;
  taosHashPut(pHash, key, strlen(key), &dummy_val, sizeof(uint8_t));
wmmhello's avatar
wmmhello 已提交
944

wmmhello's avatar
wmmhello 已提交
945
  return false;
wmmhello's avatar
wmmhello 已提交
946 947
}

wmmhello's avatar
wmmhello 已提交
948
static int32_t smlParseString(const char* sql, SSmlLineInfo *elements, SSmlMsgBuf *msg){
wmmhello's avatar
wmmhello 已提交
949 950 951 952 953 954 955 956
  if(!sql) return TSDB_CODE_SML_INVALID_DATA;
  while (*sql != '\0') {           // jump the space at the begining
    if(*sql != SPACE) {
      elements->measure = sql;
      break;
    }
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
957 958 959 960
  if (!elements->measure || *sql == COMMA) {
    smlBuildInvalidDataMsg(msg, "invalid data", sql);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982

  // parse measure and tag
  while (*sql != '\0') {
    if (elements->measureLen == 0 && *sql == COMMA && *(sql - 1) != SLASH) {  // find the first comma
      elements->measureLen = sql - elements->measure;
      sql++;
      elements->tags = sql;
      continue;
    }

    if (*sql == SPACE && *(sql - 1) != SLASH) {   // find the first space
      if (elements->measureLen == 0) {
        elements->measureLen = sql - elements->measure;
        elements->tags = sql;
      }
      elements->tagsLen = sql - elements->tags;
      elements->measureTagsLen = sql - elements->measure;
      break;
    }

    sql++;
  }
wmmhello's avatar
wmmhello 已提交
983 984 985 986 987 988 989
  if(elements->tagsLen == 0){     // measure, cols1=a         measure cols1=a
    elements->measureTagsLen = elements->measureLen;
  }
  if(elements->measureLen == 0) {
    smlBuildInvalidDataMsg(msg, "invalid measure", elements->measure);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
990 991 992 993 994 995 996 997 998

  // parse cols
  while (*sql != '\0') {
    if(*sql != SPACE) {
      elements->cols = sql;
      break;
    }
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
999 1000 1001
  if(!elements->cols) {
    smlBuildInvalidDataMsg(msg, "invalid columns", elements->cols);
    return TSDB_CODE_SML_INVALID_DATA;
1002
  }
wmmhello's avatar
wmmhello 已提交
1003

wmmhello's avatar
wmmhello 已提交
1004
  bool isInQuote = false;
wmmhello's avatar
wmmhello 已提交
1005
  while (*sql != '\0') {
wmmhello's avatar
wmmhello 已提交
1006 1007 1008 1009
    if(*sql == QUOTE && *(sql - 1) != SLASH){
      isInQuote = !isInQuote;
    }
    if(!isInQuote && *sql == SPACE && *(sql - 1) != SLASH) {
wmmhello's avatar
wmmhello 已提交
1010 1011 1012 1013
      break;
    }
    sql++;
  }
1014 1015 1016 1017
  if(isInQuote){
    smlBuildInvalidDataMsg(msg, "only one quote", elements->cols);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
1018
  elements->colsLen = sql - elements->cols;
wmmhello's avatar
wmmhello 已提交
1019

wmmhello's avatar
wmmhello 已提交
1020
  // parse ts,ts can be empty
wmmhello's avatar
wmmhello 已提交
1021
  while (*sql != '\0') {
1022
    if(*sql != SPACE && elements->timestamp == NULL) {
wmmhello's avatar
wmmhello 已提交
1023
      elements->timestamp = sql;
1024 1025
    }
    if(*sql == SPACE && elements->timestamp != NULL){
wmmhello's avatar
wmmhello 已提交
1026 1027 1028 1029
      break;
    }
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
1030 1031 1032
  if(elements->timestamp){
    elements->timestampLen = sql - elements->timestamp;
  }
wmmhello's avatar
wmmhello 已提交
1033 1034 1035 1036

  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
1037
static int32_t smlParseCols(const char* data, int32_t len, SArray *cols, bool isTag, SSmlMsgBuf *msg){
wmmhello's avatar
wmmhello 已提交
1038 1039 1040
  if(isTag && len == 0){
    SSmlKv *kv = taosMemoryCalloc(sizeof(SSmlKv), 1);
    kv->key = TAG;
wmmhello's avatar
wmmhello 已提交
1041
    kv->keyLen = strlen(TAG);
wmmhello's avatar
wmmhello 已提交
1042
    kv->value = TAG;
wmmhello's avatar
wmmhello 已提交
1043
    kv->valueLen = strlen(TAG);
wmmhello's avatar
wmmhello 已提交
1044 1045
    kv->type = TSDB_DATA_TYPE_NCHAR;
    if(cols) taosArrayPush(cols, &kv);
wmmhello's avatar
wmmhello 已提交
1046
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1047 1048
  }

wmmhello's avatar
wmmhello 已提交
1049
  for(int i = 0; i < len; i++){
wmmhello's avatar
wmmhello 已提交
1050
    // parse key
wmmhello's avatar
wmmhello 已提交
1051 1052 1053 1054 1055 1056 1057 1058 1059
    const char *key = data + i;
    int32_t keyLen = 0;
    while(i < len){
      if(data[i] == EQUAL && i > 0 && data[i-1] != SLASH){
        keyLen = data + i - key;
        break;
      }
      i++;
    }
wmmhello's avatar
wmmhello 已提交
1060
    if(keyLen == 0 || keyLen >= TSDB_COL_NAME_LEN){
wmmhello's avatar
wmmhello 已提交
1061
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1062 1063 1064
      return TSDB_CODE_SML_INVALID_DATA;
    }

wmmhello's avatar
wmmhello 已提交
1065
    // parse value
wmmhello's avatar
wmmhello 已提交
1066 1067
    i++;
    const char *value = data + i;
wmmhello's avatar
wmmhello 已提交
1068
    bool isInQuote = false;
wmmhello's avatar
wmmhello 已提交
1069
    while(i < len){
wmmhello's avatar
wmmhello 已提交
1070 1071 1072 1073
      if(data[i] == QUOTE && data[i-1] != SLASH){
        isInQuote = !isInQuote;
      }
      if(!isInQuote && data[i] == COMMA && i > 0 && data[i-1] != SLASH){
wmmhello's avatar
wmmhello 已提交
1074 1075 1076 1077
        break;
      }
      i++;
    }
wmmhello's avatar
wmmhello 已提交
1078 1079 1080 1081
    if(isInQuote){
      smlBuildInvalidDataMsg(msg, "only one quote", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }
wmmhello's avatar
wmmhello 已提交
1082
    int32_t valueLen = data + i - value;
wmmhello's avatar
wmmhello 已提交
1083
    if(valueLen == 0){
wmmhello's avatar
wmmhello 已提交
1084
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1085 1086
      return TSDB_CODE_SML_INVALID_DATA;
    }
wmmhello's avatar
wmmhello 已提交
1087 1088

    // add kv to SSmlKv
wmmhello's avatar
wmmhello 已提交
1089
    SSmlKv *kv = taosMemoryCalloc(sizeof(SSmlKv), 1);
wmmhello's avatar
wmmhello 已提交
1090 1091 1092 1093
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
    kv->valueLen = valueLen;
wmmhello's avatar
wmmhello 已提交
1094 1095
    if(isTag){
      kv->type = TSDB_DATA_TYPE_NCHAR;
wmmhello's avatar
wmmhello 已提交
1096
    }else{
wmmhello's avatar
wmmhello 已提交
1097
      if(!smlParseValue(kv, msg)){
wmmhello's avatar
wmmhello 已提交
1098 1099
        return TSDB_CODE_SML_INVALID_DATA;
      }
wmmhello's avatar
wmmhello 已提交
1100
    }
wmmhello's avatar
wmmhello 已提交
1101

wmmhello's avatar
wmmhello 已提交
1102 1103
    if(cols) taosArrayPush(cols, &kv);
  }
wmmhello's avatar
wmmhello 已提交
1104

wmmhello's avatar
wmmhello 已提交
1105 1106 1107
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
1108 1109 1110 1111 1112 1113
static int64_t smlGetTimeValue(const char *value, int32_t len, int8_t type) {
  char *endPtr = NULL;
  double ts = (double)strtoll(value, &endPtr, 10);
  if(value + len != endPtr){
    return -1;
  }
wmmhello's avatar
wmmhello 已提交
1114 1115 1116
  switch (type) {
    case TSDB_TIME_PRECISION_HOURS:
      ts *= (3600 * 1e9);
wmmhello's avatar
wmmhello 已提交
1117
      break;
wmmhello's avatar
wmmhello 已提交
1118 1119
    case TSDB_TIME_PRECISION_MINUTES:
      ts *= (60 * 1e9);
wmmhello's avatar
wmmhello 已提交
1120
      break;
wmmhello's avatar
wmmhello 已提交
1121 1122
    case TSDB_TIME_PRECISION_SECONDS:
      ts *= (1e9);
wmmhello's avatar
wmmhello 已提交
1123
      break;
wmmhello's avatar
wmmhello 已提交
1124
    case TSDB_TIME_PRECISION_MILLI:
wmmhello's avatar
wmmhello 已提交
1125
      ts *= (1e6);
wmmhello's avatar
wmmhello 已提交
1126
      break;
wmmhello's avatar
wmmhello 已提交
1127
    case TSDB_TIME_PRECISION_MICRO:
wmmhello's avatar
wmmhello 已提交
1128 1129
      ts *= (1e3);
      break;
wmmhello's avatar
wmmhello 已提交
1130 1131
    case TSDB_TIME_PRECISION_NANO:
      break;
wmmhello's avatar
wmmhello 已提交
1132 1133
    default:
      ASSERT(0);
wmmhello's avatar
wmmhello 已提交
1134 1135 1136 1137
  }
  if(ts > (double)INT64_MAX || ts < 0){
    return -1;
  }
wmmhello's avatar
wmmhello 已提交
1138 1139

  return (int64_t)ts;
wmmhello's avatar
wmmhello 已提交
1140 1141
}

wmmhello's avatar
wmmhello 已提交
1142
static int64_t smlGetTimeNow(int8_t precision) {
wmmhello's avatar
wmmhello 已提交
1143 1144 1145 1146 1147 1148 1149
  switch (precision) {
    case TSDB_TIME_PRECISION_HOURS:
      return taosGetTimestampMs()/1000/3600;
    case TSDB_TIME_PRECISION_MINUTES:
      return taosGetTimestampMs()/1000/60;
    case TSDB_TIME_PRECISION_SECONDS:
      return taosGetTimestampMs()/1000;
wmmhello's avatar
wmmhello 已提交
1150 1151 1152
    case TSDB_TIME_PRECISION_MILLI:
    case TSDB_TIME_PRECISION_MICRO:
    case TSDB_TIME_PRECISION_NANO:
wmmhello's avatar
wmmhello 已提交
1153
      return taosGetTimestamp(precision);
wmmhello's avatar
wmmhello 已提交
1154 1155
    default:
      ASSERT(0);
wmmhello's avatar
wmmhello 已提交
1156 1157 1158
  }
}

wmmhello's avatar
wmmhello 已提交
1159
static int8_t smlGetTsTypeByLen(int32_t len) {
wmmhello's avatar
wmmhello 已提交
1160 1161 1162 1163 1164
  if (len == TSDB_TIME_PRECISION_SEC_DIGITS) {
    return TSDB_TIME_PRECISION_SECONDS;
  } else if (len == TSDB_TIME_PRECISION_MILLI_DIGITS) {
    return TSDB_TIME_PRECISION_MILLI_DIGITS;
  } else {
wmmhello's avatar
wmmhello 已提交
1165
    return -1;
wmmhello's avatar
wmmhello 已提交
1166 1167 1168
  }
}

wmmhello's avatar
wmmhello 已提交
1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225
static int8_t smlGetTsTypeByPrecision(int8_t precision) {
  switch (precision) {
    case TSDB_SML_TIMESTAMP_HOURS:
      return TSDB_TIME_PRECISION_HOURS;
    case TSDB_SML_TIMESTAMP_MILLI_SECONDS:
      return TSDB_TIME_PRECISION_MILLI;
    case TSDB_SML_TIMESTAMP_NANO_SECONDS:
    case TSDB_SML_TIMESTAMP_NOT_CONFIGURED:
      return TSDB_TIME_PRECISION_NANO;
    case TSDB_SML_TIMESTAMP_MICRO_SECONDS:
      return TSDB_TIME_PRECISION_MICRO;
    case TSDB_SML_TIMESTAMP_SECONDS:
      return TSDB_TIME_PRECISION_SECONDS;
    case TSDB_SML_TIMESTAMP_MINUTES:
      return TSDB_TIME_PRECISION_MINUTES;
    default:
      return -1;
  }
}

static int64_t smlParseInfluxTime(SSmlHandle* info, const char* data, int32_t len){
  int8_t tsType = smlGetTsTypeByPrecision(info->precision);
  if (tsType == -1) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp precision", NULL);
    return -1;
  }
  if(!data){
    return smlGetTimeNow(tsType);
  }

  int64_t ts = smlGetTimeValue(data, len, tsType);
  if(ts == -1){
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
  }
  return ts;
}

static int64_t smlParseOpenTsdbTime(SSmlHandle* info, const char* data, int32_t len){
  if(!data){
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp can not be null", NULL);
    return -1;
  }
  int8_t tsType = smlGetTsTypeByLen(len);
  if (tsType == -1) {
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", data);
    return -1;
  }
  int64_t ts = smlGetTimeValue(data, len, tsType);
  if(ts == -1){
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
  }
  return ts;
}

static int32_t smlParseTS(SSmlHandle* info, const char* data, int32_t len, SArray *cols){
wmmhello's avatar
wmmhello 已提交
1226
  int64_t ts = 0;
wmmhello's avatar
wmmhello 已提交
1227 1228
  if(info->protocol == TSDB_SML_LINE_PROTOCOL){
    ts = smlParseInfluxTime(info, data, len);
wmmhello's avatar
wmmhello 已提交
1229
  }else{
wmmhello's avatar
wmmhello 已提交
1230
    ts = smlParseOpenTsdbTime(info, data, len);
wmmhello's avatar
wmmhello 已提交
1231
  }
wmmhello's avatar
wmmhello 已提交
1232
  if(ts == -1)  return TSDB_CODE_TSC_INVALID_TIME_STAMP;
wmmhello's avatar
wmmhello 已提交
1233

wmmhello's avatar
wmmhello 已提交
1234
  // add ts to
wmmhello's avatar
wmmhello 已提交
1235
  SSmlKv *kv = taosMemoryCalloc(sizeof(SSmlKv), 1);
wmmhello's avatar
wmmhello 已提交
1236 1237 1238 1239 1240
  if(!kv){
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  kv->key = TS;
wmmhello's avatar
wmmhello 已提交
1241
  kv->keyLen = strlen(kv->key);
wmmhello's avatar
wmmhello 已提交
1242
  kv->i = ts;
wmmhello's avatar
wmmhello 已提交
1243
  kv->type = TSDB_DATA_TYPE_TIMESTAMP;
wmmhello's avatar
wmmhello 已提交
1244
  kv->length = (int16_t)tDataTypes[kv->type].bytes;
wmmhello's avatar
wmmhello 已提交
1245
  if(cols) taosArrayPush(cols, &kv);
wmmhello's avatar
wmmhello 已提交
1246 1247 1248
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
1249
//static int32_t parseSmlCols(const char* data, SArray *cols){
wmmhello's avatar
wmmhello 已提交
1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 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
//  while(*data != '\0'){
//    if(*data == EQUAL) return TSDB_CODE_SML_INVALID_DATA;
//    const char *key = data;
//    int32_t keyLen = 0;
//    while(*data != '\0'){
//      if(*data == EQUAL && *(data-1) != SLASH){
//        keyLen = data - key;
//        data ++;
//        break;
//      }
//      data++;
//    }
//    if(keyLen == 0){
//      return TSDB_CODE_SML_INVALID_DATA;
//    }
//
//    if(*data == COMMA) return TSDB_CODE_SML_INVALID_DATA;
//    const char *value = data;
//    int32_t valueLen = 0;
//    while(*data != '\0'){
//      if(*data == COMMA && *(data-1) != SLASH){
//        valueLen = data - value;
//        data ++;
//        break;
//      }
//      data++;
//    }
//    if(valueLen == 0){
//      return TSDB_CODE_SML_INVALID_DATA;
//    }
//
//    TAOS_SML_KV *kv = taosMemoryCalloc(sizeof(TAOS_SML_KV), 1);
//    kv->key = key;
//    kv->keyLen = keyLen;
//    kv->value = value;
//    kv->valueLen = valueLen;
//    kv->type = TSDB_DATA_TYPE_NCHAR;
//    if(cols) taosArrayPush(cols, &kv);
//  }
//  return TSDB_CODE_SUCCESS;
//}

wmmhello's avatar
wmmhello 已提交
1292
static bool smlUpdateMeta(SSmlSTableMeta* tableMeta, SArray *tags, SArray *cols, SSmlMsgBuf *msg){
wmmhello's avatar
wmmhello 已提交
1293 1294
  if(tags){
    for (int i = 0; i < taosArrayGetSize(tags); ++i) {
wmmhello's avatar
wmmhello 已提交
1295
      SSmlKv *kv = taosArrayGetP(tags, i);
wmmhello's avatar
wmmhello 已提交
1296 1297
      ASSERT(kv->type == TSDB_DATA_TYPE_NCHAR);

wmmhello's avatar
wmmhello 已提交
1298
      SSmlKv **value = taosHashGet(tableMeta->tagHash, kv->key, kv->keyLen);
wmmhello's avatar
wmmhello 已提交
1299
      if(value){
wmmhello's avatar
wmmhello 已提交
1300 1301 1302
        ASSERT((*value)->type == TSDB_DATA_TYPE_NCHAR);
        if(kv->valueLen > (*value)->valueLen){    // tags type is nchar
          *value = kv;
wmmhello's avatar
wmmhello 已提交
1303 1304 1305 1306 1307 1308 1309 1310 1311
        }
      }else{
        taosHashPut(tableMeta->tagHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
      }
    }
  }

  if(cols){
    for (int i = 1; i < taosArrayGetSize(cols); ++i) {  //jump timestamp
wmmhello's avatar
wmmhello 已提交
1312 1313
      SSmlKv *kv = taosArrayGetP(cols, i);
      SSmlKv **value = taosHashGet(tableMeta->fieldHash, kv->key, kv->keyLen);
wmmhello's avatar
wmmhello 已提交
1314 1315
      if(value){
        if(kv->type != (*value)->type){
wmmhello's avatar
wmmhello 已提交
1316
          smlBuildInvalidDataMsg(msg, "the type is not the same like before", kv->key);
wmmhello's avatar
wmmhello 已提交
1317 1318 1319 1320 1321 1322 1323
          return false;
        }else{
          if(IS_VAR_DATA_TYPE(kv->type)){     // update string len, if bigger
            if(kv->valueLen > (*value)->valueLen){
              *value = kv;
            }
          }
wmmhello's avatar
wmmhello 已提交
1324 1325 1326 1327 1328 1329
        }
      }else{
        taosHashPut(tableMeta->fieldHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
      }
    }
  }
wmmhello's avatar
wmmhello 已提交
1330
  return true;
wmmhello's avatar
wmmhello 已提交
1331 1332
}

wmmhello's avatar
wmmhello 已提交
1333
static void smlInsertMeta(SSmlSTableMeta* tableMeta, SArray *tags, SArray *cols){
wmmhello's avatar
wmmhello 已提交
1334 1335
  if(tags){
    for (int i = 0; i < taosArrayGetSize(tags); ++i) {
wmmhello's avatar
wmmhello 已提交
1336
      SSmlKv *kv = taosArrayGetP(tags, i);
wmmhello's avatar
wmmhello 已提交
1337 1338 1339 1340 1341 1342
      taosHashPut(tableMeta->tagHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
    }
  }

  if(cols){
    for (int i = 0; i < taosArrayGetSize(cols); ++i) {
wmmhello's avatar
wmmhello 已提交
1343
      SSmlKv *kv = taosArrayGetP(cols, i);
wmmhello's avatar
wmmhello 已提交
1344 1345 1346 1347 1348
      taosHashPut(tableMeta->fieldHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
    }
  }
}

wmmhello's avatar
wmmhello 已提交
1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367
static SSmlTableInfo* smlBuildTableInfo(bool format){
  SSmlTableInfo *tag = taosMemoryCalloc(sizeof(SSmlTableInfo), 1);
  if(!tag){
    return NULL;
  }

  if(format){
    tag->colsFormat = taosArrayInit(16, POINTER_BYTES);
    if (tag->colsFormat == NULL) {
      uError("SML:smlParseLine failed to allocate memory");
      goto cleanup;
    }
  }else{
    tag->cols = taosArrayInit(16, POINTER_BYTES);
    if (tag->cols == NULL) {
      uError("SML:smlParseLine failed to allocate memory");
      goto cleanup;
    }

1368
    tag->columnsHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
wmmhello's avatar
wmmhello 已提交
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
    if (tag->columnsHash == NULL) {
      uError("SML:smlParseLine failed to allocate memory");
      goto cleanup;
    }
  }

  tag->tags = taosArrayInit(16, POINTER_BYTES);
  if (tag->tags == NULL) {
    uError("SML:smlParseLine failed to allocate memory");
    goto cleanup;
  }
  return tag;

cleanup:
  taosMemoryFreeClear(tag);
  return NULL;
}

static void smlDestroyBuildTableInfo(SSmlTableInfo *tag, bool format){
  if(format){
    taosArrayDestroy(tag->colsFormat);
  }else{
    tag->cols = taosArrayInit(16, POINTER_BYTES);
    for(size_t i = 0; i < taosArrayGetSize(tag->cols); i++){
      SHashObj *kvHash = taosArrayGetP(tag->cols, i);
      void** p1 = taosHashIterate(kvHash, NULL);
      while (p1) {
        SSmlKv* kv = *p1;
        taosMemoryFreeClear(kv);
        p1 = taosHashIterate(kvHash, p1);
      }
      taosHashCleanup(kvHash);
    }
    taosHashCleanup(tag->columnsHash);
  }
  taosArrayDestroy(tag->tags);
  taosMemoryFreeClear(tag);
}

static int32_t smlDealCols(SSmlTableInfo* oneTable, bool dataFormat, SArray *cols){
  if(dataFormat){
    taosArrayPush(oneTable->colsFormat, &cols);
  }else{
1412
    SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
wmmhello's avatar
wmmhello 已提交
1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427
    if(!kvHash){
      uError("SML:smlDealCols failed to allocate memory");
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    for(size_t i = 0; i < taosArrayGetSize(cols); i++){
      SSmlKv *kv = taosArrayGetP(cols, i);
      taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);   // todo key need escape, like \=, because find by schema name later

      if(taosHashGet(oneTable->columnsHash, kv->key, kv->keyLen) != NULL){
        continue;
      }
      taosHashPut(oneTable->columnsHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
    }
    taosArrayPush(oneTable->cols, &kvHash);
  }
1428
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1429 1430 1431 1432 1433 1434 1435
}

static SSmlSTableMeta* smlBuildSTableMeta(){
  SSmlSTableMeta* meta = taosMemoryCalloc(sizeof(SSmlSTableMeta), 1);
  if(!meta){
    return NULL;
  }
1436
  meta->tagHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
wmmhello's avatar
wmmhello 已提交
1437 1438 1439 1440 1441
  if (meta->tagHash == NULL) {
    uError("SML:smlBuildSTableMeta failed to allocate memory");
    goto cleanup;
  }

1442
  meta->fieldHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
wmmhello's avatar
wmmhello 已提交
1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462
  if (meta->fieldHash == NULL) {
    uError("SML:smlBuildSTableMeta failed to allocate memory");
    goto cleanup;
  }
  return meta;

cleanup:
  taosMemoryFreeClear(meta);
  return NULL;
}

static void smlDestroySTableMeta(SSmlSTableMeta *meta){
  taosHashCleanup(meta->tagHash);
  taosHashCleanup(meta->fieldHash);
  taosMemoryFree(meta->tableMeta);
}

static int32_t smlParseLine(SSmlHandle* info, const char* sql) {
  SSmlLineInfo elements = {0};
  int ret = smlParseString(sql, &elements, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
1463
  if(ret != TSDB_CODE_SUCCESS){
wmmhello's avatar
wmmhello 已提交
1464
    uError("SML:0x%"PRIx64" smlParseString failed", info->id);
wmmhello's avatar
wmmhello 已提交
1465 1466 1467 1468 1469
    return ret;
  }

  SArray *cols = taosArrayInit(16, POINTER_BYTES);
  if (cols == NULL) {
wmmhello's avatar
wmmhello 已提交
1470
    uError("SML:0x%"PRIx64" smlParseLine failed to allocate memory", info->id);
wmmhello's avatar
wmmhello 已提交
1471 1472
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
wmmhello's avatar
wmmhello 已提交
1473

wmmhello's avatar
wmmhello 已提交
1474
  ret = smlParseTS(info, elements.timestamp, elements.timestampLen, cols);
wmmhello's avatar
wmmhello 已提交
1475
  if(ret != TSDB_CODE_SUCCESS){
wmmhello's avatar
wmmhello 已提交
1476
    uError("SML:0x%"PRIx64" smlParseTS failed", info->id);
wmmhello's avatar
wmmhello 已提交
1477 1478
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
1479
  ret = smlParseCols(elements.cols, elements.colsLen, cols, false, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
1480
  if(ret != TSDB_CODE_SUCCESS){
wmmhello's avatar
wmmhello 已提交
1481
    uError("SML:0x%"PRIx64" smlParseCols parse cloums fields failed", info->id);
wmmhello's avatar
wmmhello 已提交
1482 1483
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
1484
  if(taosArrayGetSize(cols) > TSDB_MAX_COLUMNS){
wmmhello's avatar
wmmhello 已提交
1485
    smlBuildInvalidDataMsg(&info->msgBuf, "too many columns than 4096", NULL);
wmmhello's avatar
wmmhello 已提交
1486 1487
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
1488

wmmhello's avatar
wmmhello 已提交
1489
  SSmlTableInfo **oneTable = taosHashGet(info->childTables, elements.measure, elements.measureTagsLen);
wmmhello's avatar
wmmhello 已提交
1490 1491 1492
  if(oneTable){
    SSmlSTableMeta** tableMeta = taosHashGet(info->superTables, elements.measure, elements.measureLen);
    ASSERT(tableMeta);
wmmhello's avatar
wmmhello 已提交
1493
    ret = smlUpdateMeta(*tableMeta, NULL, cols, &info->msgBuf);    // update meta cols
wmmhello's avatar
wmmhello 已提交
1494
    if(!ret){
wmmhello's avatar
wmmhello 已提交
1495
      uError("SML:0x%"PRIx64" smlUpdateMeta cols failed", info->id);
wmmhello's avatar
wmmhello 已提交
1496 1497
      return TSDB_CODE_SML_INVALID_DATA;
    }
wmmhello's avatar
wmmhello 已提交
1498 1499 1500 1501
    ret = smlDealCols(*oneTable, info->dataFormat, cols);
    if(ret != TSDB_CODE_SUCCESS){
      return ret;
    }
wmmhello's avatar
wmmhello 已提交
1502
  }else{
wmmhello's avatar
wmmhello 已提交
1503
    SSmlTableInfo *tag = smlBuildTableInfo(info->dataFormat);
wmmhello's avatar
wmmhello 已提交
1504
    if(!tag){
wmmhello's avatar
wmmhello 已提交
1505 1506
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
wmmhello's avatar
wmmhello 已提交
1507 1508 1509
    ret = smlDealCols(tag, info->dataFormat, cols);
    if(ret != TSDB_CODE_SUCCESS){
      return ret;
wmmhello's avatar
wmmhello 已提交
1510 1511
    }

wmmhello's avatar
wmmhello 已提交
1512
    ret = smlParseCols(elements.tags, elements.tagsLen, tag->tags, true, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
1513
    if(ret != TSDB_CODE_SUCCESS){
wmmhello's avatar
wmmhello 已提交
1514
      uError("SML:0x%"PRIx64" smlParseCols parse tag fields failed", info->id);
wmmhello's avatar
wmmhello 已提交
1515 1516 1517
      return ret;
    }

wmmhello's avatar
wmmhello 已提交
1518
    if(taosArrayGetSize(tag->tags) > TSDB_MAX_TAGS){
wmmhello's avatar
wmmhello 已提交
1519
      smlBuildInvalidDataMsg(&info->msgBuf, "too many tags than 128", NULL);
wmmhello's avatar
wmmhello 已提交
1520 1521 1522 1523 1524
      return TSDB_CODE_SML_INVALID_DATA;
    }

    tag->sTableName = elements.measure;
    tag->sTableNameLen = elements.measureLen;
wmmhello's avatar
wmmhello 已提交
1525 1526
    smlBuildChildTableName(tag);
    uDebug("SML:0x%"PRIx64" child table name: %s", info->id, tag->childTableName);
wmmhello's avatar
wmmhello 已提交
1527

wmmhello's avatar
wmmhello 已提交
1528 1529
    SSmlSTableMeta** tableMeta = taosHashGet(info->superTables, elements.measure, elements.measureLen);
    if(tableMeta){  // update meta
wmmhello's avatar
wmmhello 已提交
1530
      ret = smlUpdateMeta(*tableMeta, tag->tags, cols, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
1531
      if(!ret){
wmmhello's avatar
wmmhello 已提交
1532
        uError("SML:0x%"PRIx64" smlUpdateMeta failed", info->id);
wmmhello's avatar
wmmhello 已提交
1533 1534
        return TSDB_CODE_SML_INVALID_DATA;
      }
wmmhello's avatar
wmmhello 已提交
1535
    }else{
wmmhello's avatar
wmmhello 已提交
1536 1537
      SSmlSTableMeta *meta = smlBuildSTableMeta();
      smlInsertMeta(meta, tag->tags, cols);
wmmhello's avatar
wmmhello 已提交
1538 1539 1540 1541 1542 1543 1544 1545
      taosHashPut(info->superTables, elements.measure, elements.measureLen, &meta, POINTER_BYTES);
    }

    taosHashPut(info->childTables, elements.measure, elements.measureTagsLen, &tag, POINTER_BYTES);
  }
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
1546
static void smlDestroyInfo(SSmlHandle* info){
wmmhello's avatar
wmmhello 已提交
1547 1548
  if(!info) return;
  qDestroyQuery(info->pQuery);
wmmhello's avatar
wmmhello 已提交
1549 1550 1551 1552 1553 1554 1555 1556 1557
  smlDestroyHandle(info->exec);

  // destroy info->childTables
  void** p1 = taosHashIterate(info->childTables, NULL);
  while (p1) {
    SSmlTableInfo* oneTable = *p1;
    smlDestroyBuildTableInfo(oneTable, info->dataFormat);
    p1 = taosHashIterate(info->childTables, p1);
  }
wmmhello's avatar
wmmhello 已提交
1558
  taosHashCleanup(info->childTables);
wmmhello's avatar
wmmhello 已提交
1559 1560 1561 1562 1563 1564 1565 1566

  // destroy info->superTables
  p1 = taosHashIterate(info->superTables, NULL);
  while (p1) {
    SSmlSTableMeta* oneTable = *p1;
    smlDestroySTableMeta(oneTable);
    p1 = taosHashIterate(info->superTables, p1);
  }
wmmhello's avatar
wmmhello 已提交
1567
  taosHashCleanup(info->superTables);
wmmhello's avatar
wmmhello 已提交
1568 1569

  // destroy info->pVgHash
wmmhello's avatar
wmmhello 已提交
1570
  taosHashCleanup(info->pVgHash);
wmmhello's avatar
wmmhello 已提交
1571 1572

  taosMemoryFreeClear(info);
wmmhello's avatar
wmmhello 已提交
1573
}
wmmhello's avatar
wmmhello 已提交
1574 1575

static SSmlHandle* smlBuildSmlInfo(TAOS* taos, SRequestObj* request, SMLProtocolType protocol, int8_t precision, bool dataFormat){
1576
  int32_t code = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1577
  SSmlHandle* info = taosMemoryMalloc(sizeof(SSmlHandle));
wmmhello's avatar
wmmhello 已提交
1578 1579 1580
  if (NULL == info) {
    return NULL;
  }
wmmhello's avatar
wmmhello 已提交
1581
  info->id          = smlGenId();
wmmhello's avatar
wmmhello 已提交
1582

wmmhello's avatar
wmmhello 已提交
1583
  info->pQuery      = taosMemoryCalloc(1, sizeof(SQuery));
wmmhello's avatar
wmmhello 已提交
1584
  if (NULL == info->pQuery) {
wmmhello's avatar
wmmhello 已提交
1585
    uError("SML:0x%"PRIx64" create info->pQuery error", info->id);
wmmhello's avatar
wmmhello 已提交
1586 1587
    goto cleanup;
  }
wmmhello's avatar
wmmhello 已提交
1588
  info->pQuery->execMode      = QUERY_EXEC_MODE_SCHEDULE;
wmmhello's avatar
wmmhello 已提交
1589
  info->pQuery->haveResultSet = false;
wmmhello's avatar
wmmhello 已提交
1590 1591 1592 1593 1594 1595
  info->pQuery->msgType       = TDMT_VND_SUBMIT;
  info->pQuery->pRoot         = (SNode*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT);
  if(NULL == info->pQuery->pRoot){
    uError("SML:0x%"PRIx64" create info->pQuery->pRoot error", info->id);
    goto cleanup;
  }
wmmhello's avatar
wmmhello 已提交
1596 1597
  ((SVnodeModifOpStmt*)(info->pQuery->pRoot))->payloadType = PAYLOAD_TYPE_KV;

wmmhello's avatar
wmmhello 已提交
1598
  info->taos        = taos;
1599
  code = catalogGetHandle(info->taos->pAppInfo->clusterId, &info->pCatalog);
wmmhello's avatar
wmmhello 已提交
1600 1601
  if(code != TSDB_CODE_SUCCESS){
    uError("SML:0x%"PRIx64" get catalog error %d", info->id, code);
wmmhello's avatar
wmmhello 已提交
1602 1603 1604
    goto cleanup;
  }

wmmhello's avatar
wmmhello 已提交
1605 1606 1607 1608 1609 1610 1611 1612
  info->precision   = precision;
  info->protocol    = protocol;
  info->dataFormat  = dataFormat;
  info->pRequest    = request;
  info->msgBuf.buf  = info->pRequest->msgBuf;
  info->msgBuf.len  = ERROR_MSG_BUF_DEFAULT_SIZE;

  info->exec        = smlInitHandle(info->pQuery);
1613 1614 1615
  info->childTables = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  info->superTables = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  info->pVgHash     = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
wmmhello's avatar
wmmhello 已提交
1616

wmmhello's avatar
wmmhello 已提交
1617 1618 1619 1620 1621
  if(NULL == info->exec || NULL == info->childTables
      || NULL == info->superTables || NULL == info->pVgHash){
    uError("SML:0x%"PRIx64" create info failed", info->id);
    goto cleanup;
  }
wmmhello's avatar
wmmhello 已提交
1622

wmmhello's avatar
wmmhello 已提交
1623
  return info;
wmmhello's avatar
wmmhello 已提交
1624 1625 1626 1627
cleanup:
  smlDestroyInfo(info);
  return NULL;
}
wmmhello's avatar
wmmhello 已提交
1628
static int32_t smlInsertData(SSmlHandle* info) {
wmmhello's avatar
wmmhello 已提交
1629 1630
  int32_t code = TSDB_CODE_SUCCESS;

wmmhello's avatar
wmmhello 已提交
1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649
  SSmlTableInfo** oneTable = taosHashIterate(info->childTables, NULL);
  while (oneTable) {
    SSmlTableInfo* tableData = *oneTable;

    SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
    strcpy(pName.dbname, info->pRequest->pDb);
    memcpy(pName.tname, tableData->childTableName, strlen(tableData->childTableName));
    SEpSet ep = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
    SVgroupInfo vg;
    code = catalogGetTableHashVgroup(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, &vg);
    if (code != 0) {
      uError("SML:0x%"PRIx64" catalogGetTableHashVgroup failed. table name: %s", info->id, tableData->childTableName);
      return code;
    }
    taosHashPut(info->pVgHash, (const char*)&vg.vgId, sizeof(vg.vgId), (char*)&vg, sizeof(vg));

    SSmlSTableMeta** pMeta = taosHashGet(info->superTables, tableData->sTableName, tableData->sTableNameLen);
    ASSERT (NULL != pMeta && NULL != *pMeta);

1650
    // use tablemeta of stable to save vgid and uid of child table
wmmhello's avatar
wmmhello 已提交
1651 1652 1653 1654
    (*pMeta)->tableMeta->vgId = vg.vgId;
    (*pMeta)->tableMeta->uid = tableData->uid; // one table merge data block together according uid

    code = smlBindData(info->exec, tableData->tags, tableData->colsFormat, tableData->columnsHash,
1655
                       tableData->cols, info->dataFormat, (*pMeta)->tableMeta, tableData->childTableName, info->msgBuf.buf, info->msgBuf.len);
wmmhello's avatar
wmmhello 已提交
1656 1657 1658 1659
    if(code != TSDB_CODE_SUCCESS){
      return code;
    }
    oneTable = taosHashIterate(info->childTables, oneTable);
wmmhello's avatar
wmmhello 已提交
1660
  }
wmmhello's avatar
wmmhello 已提交
1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671

  smlBuildOutput(info->exec, info->pVgHash);
  launchQueryImpl(info->pRequest, info->pQuery, TSDB_CODE_SUCCESS, true);

  info->affectedRows = taos_affected_rows(info->pRequest);
  return info->pRequest->code;
}

static int smlInsertLines(SSmlHandle *info, char* lines[], int numLines) {
  int32_t code = TSDB_CODE_SUCCESS;

wmmhello's avatar
wmmhello 已提交
1672
  if (numLines <= 0 || numLines > 65536) {
wmmhello's avatar
wmmhello 已提交
1673
    uError("SML:0x%"PRIx64" smlInsertLines numLines should be between 1 and 65536. numLines: %d", info->id, numLines);
wmmhello's avatar
wmmhello 已提交
1674
    code = TSDB_CODE_TSC_APP_ERROR;
wmmhello's avatar
wmmhello 已提交
1675 1676
    goto cleanup;
  }
wmmhello's avatar
wmmhello 已提交
1677

wmmhello's avatar
wmmhello 已提交
1678
  for (int32_t i = 0; i < numLines; ++i) {
wmmhello's avatar
wmmhello 已提交
1679
    code = smlParseLine(info, lines[i]);
wmmhello's avatar
wmmhello 已提交
1680
    if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
1681
      uError("SML:0x%"PRIx64" smlParseLine failed. line %d : %s", info->id, i, lines[i]);
wmmhello's avatar
wmmhello 已提交
1682 1683 1684
      goto cleanup;
    }
  }
wmmhello's avatar
wmmhello 已提交
1685 1686 1687 1688 1689 1690 1691 1692
  uDebug("SML:0x%"PRIx64" smlInsertLines parse success. tables %d", info->id, taosHashGetSize(info->childTables));
  uDebug("SML:0x%"PRIx64" smlInsertLines parse success. super tables %d", info->id, taosHashGetSize(info->superTables));

  code = smlModifyDBSchemas(info);
  if (code != 0) {
    uError("SML:0x%"PRIx64" smlModifyDBSchemas error : %s", info->id, tstrerror(code));
    goto cleanup;
  }
wmmhello's avatar
wmmhello 已提交
1693

wmmhello's avatar
wmmhello 已提交
1694 1695 1696
  code = smlInsertData(info);
  if (code != 0) {
    uError("SML:0x%"PRIx64" smlInsertData error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
1697 1698 1699
    goto cleanup;
  }

wmmhello's avatar
wmmhello 已提交
1700
  uDebug("SML:0x%"PRIx64" smlInsertLines finish inserting %d lines.", info->id, numLines);
wmmhello's avatar
wmmhello 已提交
1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726

cleanup:
  return code;
}

/**
 * taos_schemaless_insert() parse and insert data points into database according to
 * different protocol.
 *
 * @param $lines input array may contain multiple lines, each line indicates a data point.
 *               If protocol=2 is used input array should contain single JSON
 *               string(e.g. char *lines[] = {"$JSON_string"}). If need to insert
 *               multiple data points in JSON format, should include them in $JSON_string
 *               as a JSON array.
 * @param $numLines indicates how many data points in $lines.
 *                  If protocol = 2 is used this param will be ignored as $lines should
 *                  contain single JSON string.
 * @param $protocol indicates which protocol to use for parsing:
 *                  0 - influxDB line protocol
 *                  1 - OpenTSDB telnet line protocol
 *                  2 - OpenTSDB JSON format protocol
 * @return return zero for successful insertion. Otherwise return none-zero error code of
 *         failure reason.
 *
 */

wmmhello's avatar
wmmhello 已提交
1727
TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int protocol, int precision, bool dataFormat) {
wmmhello's avatar
wmmhello 已提交
1728
  SRequestObj* request = createRequest(taos, NULL, NULL, TSDB_SQL_INSERT);
wmmhello's avatar
wmmhello 已提交
1729
  if(!request){
1730
    return NULL;
wmmhello's avatar
wmmhello 已提交
1731 1732 1733 1734
  }

  SSmlHandle* info = smlBuildSmlInfo(taos, request, protocol, precision, dataFormat);
  if(!info){
1735
    return (TAOS_RES*)request;
wmmhello's avatar
wmmhello 已提交
1736 1737
  }

wmmhello's avatar
wmmhello 已提交
1738
  switch (protocol) {
wmmhello's avatar
wmmhello 已提交
1739
    case TSDB_SML_LINE_PROTOCOL:{
wmmhello's avatar
wmmhello 已提交
1740
      smlInsertLines(info, lines, numLines);
wmmhello's avatar
wmmhello 已提交
1741
      break;
wmmhello's avatar
wmmhello 已提交
1742
    }
wmmhello's avatar
wmmhello 已提交
1743 1744 1745 1746 1747 1748 1749 1750 1751
    case TSDB_SML_TELNET_PROTOCOL:
      //code = taos_insert_telnet_lines(taos, lines, numLines, protocol, tsType, &affected_rows);
      break;
    case TSDB_SML_JSON_PROTOCOL:
      //code = taos_insert_json_payload(taos, *lines, protocol, tsType, &affected_rows);
      break;
    default:
      break;
  }
1752
  smlDestroyInfo(info);
wmmhello's avatar
wmmhello 已提交
1753

wmmhello's avatar
wmmhello 已提交
1754 1755
end:
  return (TAOS_RES*)request;
wmmhello's avatar
wmmhello 已提交
1756
}
1757