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

H
Hongze Cheng 已提交
16
#include "tsdb.h"
H
Hongze Cheng 已提交
17

C
Cary Xu 已提交
18
// static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg);
H
Hongze Cheng 已提交
19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38

int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *pRsp) {
  SSubmitMsgIter msgIter = {0};
  SSubmitBlk    *pBlock = NULL;
  int32_t        affectedrows = 0;
  int32_t        numOfRows = 0;

  ASSERT(pTsdb->mem != NULL);

  // scan and convert
  if (tsdbScanAndConvertSubmitMsg(pTsdb, pMsg) < 0) {
    if (terrno != TSDB_CODE_TDB_TABLE_RECONFIGURE) {
      tsdbError("vgId:%d failed to insert data since %s", REPO_ID(pTsdb), tstrerror(terrno));
    }
    return -1;
  }

  // loop to insert
  tInitSubmitMsgIter(pMsg, &msgIter);
  while (true) {
H
Hongze Cheng 已提交
39
    SSubmitBlkRsp r = {0};
H
Hongze Cheng 已提交
40 41
    tGetSubmitMsgNext(&msgIter, &pBlock);
    if (pBlock == NULL) break;
H
Hongze Cheng 已提交
42
    if (tsdbInsertTableData(pTsdb, &msgIter, pBlock, &r) < 0) {
H
Hongze Cheng 已提交
43 44 45
      return -1;
    }

C
Cary Xu 已提交
46
    numOfRows += msgIter.numOfRows;
H
Hongze Cheng 已提交
47 48 49
  }

  if (pRsp != NULL) {
H
Hongze Cheng 已提交
50 51
    // pRsp->affectedRows = affectedrows;
    // pRsp->numOfRows = numOfRows;
H
Hongze Cheng 已提交
52 53 54 55 56
  }

  return 0;
}

C
Cary Xu 已提交
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
#if 0
static FORCE_INLINE int tsdbCheckRowRange(STsdb *pTsdb, STable *pTable, STSRow *row, TSKEY minKey, TSKEY maxKey,
                                          TSKEY now) {
  TSKEY rowKey = TD_ROW_KEY(row);
  if (rowKey < minKey || rowKey > maxKey) {
    tsdbError("vgId:%d table %s tid %d uid %" PRIu64 " timestamp is out of range! now %" PRId64 " minKey %" PRId64
              " maxKey %" PRId64 " row key %" PRId64,
              REPO_ID(pTsdb), TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable), now, minKey, maxKey,
              rowKey);
    terrno = TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE;
    return -1;
  }

  return 0;
}
#endif

static FORCE_INLINE int tsdbCheckRowRange(STsdb *pTsdb, tb_uid_t uid, STSRow *row, TSKEY minKey, TSKEY maxKey,
                                          TSKEY now) {
  TSKEY rowKey = TD_ROW_KEY(row);
  if (rowKey < minKey || rowKey > maxKey) {
    tsdbError("vgId:%d table uid %" PRIu64 " timestamp is out of range! now %" PRId64 " minKey %" PRId64
              " maxKey %" PRId64 " row key %" PRId64,
              REPO_ID(pTsdb), uid, now, minKey, maxKey, rowKey);
    terrno = TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE;
    return -1;
  }

  return 0;
}

H
Hongze Cheng 已提交
88
int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
H
Hongze Cheng 已提交
89 90 91 92 93 94
  ASSERT(pMsg != NULL);
  // STsdbMeta *    pMeta = pTsdb->tsdbMeta;
  SSubmitMsgIter msgIter = {0};
  SSubmitBlk    *pBlock = NULL;
  SSubmitBlkIter blkIter = {0};
  STSRow        *row = NULL;
C
Cary Xu 已提交
95
  STsdbKeepCfg  *pCfg = REPO_KEEP_CFG(pTsdb);
H
Hongze Cheng 已提交
96
  TSKEY          now = taosGetTimestamp(pCfg->precision);
97 98
  TSKEY          minKey = now - tsTickPerMin[pCfg->precision] * pCfg->keep2;
  TSKEY          maxKey = now + tsTickPerMin[pCfg->precision] * pCfg->days;
H
Hongze Cheng 已提交
99 100

  terrno = TSDB_CODE_SUCCESS;
C
Cary Xu 已提交
101 102
  // pMsg->length = htonl(pMsg->length);
  // pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
H
Hongze Cheng 已提交
103 104 105 106 107 108

  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
  while (true) {
    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
    if (pBlock == NULL) break;

C
Cary Xu 已提交
109 110 111 112 113 114
      // pBlock->uid = htobe64(pBlock->uid);
      // pBlock->suid = htobe64(pBlock->suid);
      // pBlock->sversion = htonl(pBlock->sversion);
      // pBlock->dataLen = htonl(pBlock->dataLen);
      // pBlock->schemaLen = htonl(pBlock->schemaLen);
      // pBlock->numOfRows = htons(pBlock->numOfRows);
H
Hongze Cheng 已提交
115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145

#if 0
    if (pBlock->tid <= 0 || pBlock->tid >= pMeta->maxTables) {
      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pTsdb), pBlock->uid,
                pBlock->tid);
      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
      return -1;
    }

    STable *pTable = pMeta->tables[pBlock->tid];
    if (pTable == NULL || TABLE_UID(pTable) != pBlock->uid) {
      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pTsdb), pBlock->uid,
                pBlock->tid);
      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
      return -1;
    }

    if (TABLE_TYPE(pTable) == TSDB_SUPER_TABLE) {
      tsdbError("vgId:%d invalid action trying to insert a super table %s", REPO_ID(pTsdb), TABLE_CHAR_NAME(pTable));
      terrno = TSDB_CODE_TDB_INVALID_ACTION;
      return -1;
    }

    // Check schema version and update schema if needed
    if (tsdbCheckTableSchema(pTsdb, pBlock, pTable) < 0) {
      if (terrno == TSDB_CODE_TDB_TABLE_RECONFIGURE) {
        continue;
      } else {
        return -1;
      }
    }
C
Cary Xu 已提交
146 147 148 149
#endif
    tInitSubmitBlkIter(&msgIter, pBlock, &blkIter);
    while ((row = tGetSubmitBlkNext(&blkIter)) != NULL) {
      if (tsdbCheckRowRange(pTsdb, msgIter.uid, row, minKey, maxKey, now) < 0) {
H
Hongze Cheng 已提交
150 151 152 153 154 155 156
        return -1;
      }
    }
  }

  if (terrno != TSDB_CODE_SUCCESS) return -1;
  return 0;
C
Cary Xu 已提交
157
}