From 49595291277013d1ae79a5c99d754d77c738e720 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 3 Nov 2022 19:00:13 +0800 Subject: [PATCH] fix(query): set the last key of each tablescan info --- source/dnode/vnode/src/tsdb/tsdbRead.c | 45 +- source/libs/executor/src/executorimpl.c | 5 +- tests/system-test/1-insert/delete_data.py | 614 +++++++++++----------- 3 files changed, 340 insertions(+), 324 deletions(-) diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 89d07911d9..4099dafa26 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -185,11 +185,11 @@ static int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STabl SRowMerger* pMerger, SVersionRange* pVerRange); static int32_t doMergeRowsInBuf(SIterInfo* pIter, uint64_t uid, int64_t ts, SArray* pDelList, SRowMerger* pMerger, STsdbReader* pReader); -static int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow, uint64_t uid); +static int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow, STableBlockScanInfo* pInfo); static int32_t doAppendRowFromFileBlock(SSDataBlock* pResBlock, STsdbReader* pReader, SBlockData* pBlockData, int32_t rowIndex); static void setComposedBlockFlag(STsdbReader* pReader, bool composed); -static bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order, SVersionRange* pRange); +static bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32_t order, SVersionRange* pVerRange); static int32_t doMergeMemTableMultiRows(TSDBROW* pRow, uint64_t uid, SIterInfo* pIter, SArray* pDelList, STSRow** pTSRow, STsdbReader* pReader, bool* freeTSRow); @@ -208,7 +208,6 @@ static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader); static int32_t doBuildDataBlock(STsdbReader* pReader); static TSDBKEY getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbReader* pReader); static bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo); -static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader); static bool outOfTimeWindow(int64_t ts, STimeWindow* pWindow) { return (ts > pWindow->ekey) || (ts < pWindow->skey); } @@ -1529,8 +1528,8 @@ static bool tryCopyDistinctRowFromFileBlock(STsdbReader* pReader, SBlockData* pB // opt version // 1. it is not a border point // 2. the direct next point is not an duplicated timestamp - if ((pDumpInfo->rowIndex < pDumpInfo->totalRows - 1 && pReader->order == TSDB_ORDER_ASC) || - (pDumpInfo->rowIndex > 0 && pReader->order == TSDB_ORDER_DESC)) { + bool asc = (pReader->order == TSDB_ORDER_ASC); + if ((pDumpInfo->rowIndex < pDumpInfo->totalRows - 1 && asc) || (pDumpInfo->rowIndex > 0 && (!asc))) { int32_t step = pReader->order == TSDB_ORDER_ASC ? 1 : -1; int64_t nextKey = pBlockData->aTSKEY[pDumpInfo->rowIndex + step]; @@ -1749,7 +1748,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo* return code; } - doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid); + doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo); taosMemoryFree(pTSRow); tRowMergerClear(&merge); @@ -1770,6 +1769,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader, // only last block exists if ((!mergeBlockData) || (tsLastBlock != pBlockData->aTSKEY[pDumpInfo->rowIndex])) { if (tryCopyDistinctRowFromSttBlock(&fRow, pLastBlockReader, pBlockScanInfo, tsLastBlock, pReader)) { + pBlockScanInfo->lastKey = tsLastBlock; return TSDB_CODE_SUCCESS; } else { int32_t code = tRowMergerInit(&merge, &fRow, pReader->pSchema); @@ -1786,7 +1786,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader, return code; } - doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid); + doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo); taosMemoryFree(pTSRow); tRowMergerClear(&merge); @@ -1810,7 +1810,7 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader, return code; } - doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid); + doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo); taosMemoryFree(pTSRow); tRowMergerClear(&merge); @@ -1858,7 +1858,7 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader return code; } - doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid); + doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo); taosMemoryFree(pTSRow); tRowMergerClear(&merge); @@ -2082,7 +2082,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo* return code; } - doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid); + doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo); taosMemoryFree(pTSRow); tRowMergerClear(&merge); @@ -2233,6 +2233,7 @@ int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBloc STsdbReader* pReader) { SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; if (tryCopyDistinctRowFromFileBlock(pReader, pBlockData, key, pDumpInfo)) { + pBlockScanInfo->lastKey = key; return TSDB_CODE_SUCCESS; } else { TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex); @@ -2251,7 +2252,7 @@ int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBloc return code; } - doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo->uid); + doAppendRowFromTSRow(pReader->pResBlock, pReader, pTSRow, pBlockScanInfo); taosMemoryFree(pTSRow); tRowMergerClear(&merge); @@ -2299,29 +2300,32 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) { SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter); SLastBlockReader* pLastBlockReader = pReader->status.fileIter.pLastBlockReader; + bool asc = ASCENDING_TRAVERSE(pReader->order); int64_t st = taosGetTimestampUs(); + int32_t step = asc ? 1 : -1; STableBlockScanInfo* pBlockScanInfo = NULL; if (pBlockInfo != NULL) { - pBlockScanInfo = *(STableBlockScanInfo**)taosHashGet(pReader->status.pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid)); - if (pBlockScanInfo == NULL) { + void* p = taosHashGet(pReader->status.pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid)); + if (p == NULL) { code = TSDB_CODE_INVALID_PARA; tsdbError("failed to locate the uid:%" PRIu64 " in query table uid list, total tables:%d, %s", pBlockInfo->uid, taosHashGetSize(pReader->status.pTableMap), pReader->idStr); goto _end; } + pBlockScanInfo = *(STableBlockScanInfo**) p; + SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter); TSDBKEY keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader); // it is a clean block, load it directly if (isCleanFileDataBlock(pReader, pBlockInfo, pBlock, pBlockScanInfo, keyInBuf, pLastBlockReader)) { - if (pReader->order == TSDB_ORDER_ASC || - (pReader->order == TSDB_ORDER_DESC && (!hasDataInLastBlock(pLastBlockReader)))) { + if (asc || ((!asc) && (!hasDataInLastBlock(pLastBlockReader)))) { copyBlockDataToSDataBlock(pReader, pBlockScanInfo); // record the last key value - pBlockScanInfo->lastKey = ASCENDING_TRAVERSE(pReader->order)? pBlock->maxKey.ts:pBlock->minKey.ts; + pBlockScanInfo->lastKey = asc? pBlock->maxKey.ts:pBlock->minKey.ts; goto _end; } } @@ -2331,7 +2335,6 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) { SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SBlockData* pBlockData = &pReader->status.fileBlockData; - int32_t step = ASCENDING_TRAVERSE(pReader->order) ? 1 : -1; while (1) { bool hasBlockData = false; @@ -3220,7 +3223,6 @@ int32_t doMergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pSc int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pScanInfo, int64_t ts, SRowMerger* pMerger, SVersionRange* pVerRange) { - pScanInfo->lastKey = ts; while (nextRowFromLastBlocks(pLastBlockReader, pScanInfo, pVerRange)) { int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader); if (next1 == ts) { @@ -3413,9 +3415,10 @@ int32_t tsdbGetNextRowInMem(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pR return TSDB_CODE_SUCCESS; } -int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow, uint64_t uid) { +int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow, STableBlockScanInfo* pScanInfo) { int32_t numOfRows = pBlock->info.rows; int32_t numOfCols = (int32_t)taosArrayGetSize(pBlock->pDataBlock); + int64_t uid = pScanInfo->uid; SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo; STSchema* pSchema = doGetSchemaForTSRow(pTSRow->sver, pReader, uid); @@ -3454,6 +3457,7 @@ int32_t doAppendRowFromTSRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* } pBlock->info.rows += 1; + pScanInfo->lastKey = pTSRow->ts; return TSDB_CODE_SUCCESS; } @@ -3517,7 +3521,8 @@ int32_t buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, int64_t e break; } - doAppendRowFromTSRow(pBlock, pReader, pTSRow, pBlockScanInfo->uid); + doAppendRowFromTSRow(pBlock, pReader, pTSRow, pBlockScanInfo); + if (freeTSRow) { taosMemoryFree(pTSRow); } diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 59266716e1..26abc2b90b 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -2973,7 +2973,7 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiN return pOperator; - _error: +_error: if (pInfo != NULL) { destroyAggOperatorInfo(pInfo); } @@ -3189,11 +3189,12 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* code = appendDownstream(pOperator, &downstream, 1); return pOperator; - _error: +_error: if (pInfo != NULL) { destroyFillOperatorInfo(pInfo); } + pTaskInfo->code = code; taosMemoryFreeClear(pOperator); return NULL; } diff --git a/tests/system-test/1-insert/delete_data.py b/tests/system-test/1-insert/delete_data.py index 95f7649ba1..c085d2763a 100644 --- a/tests/system-test/1-insert/delete_data.py +++ b/tests/system-test/1-insert/delete_data.py @@ -1,302 +1,312 @@ -################################################################### -# Copyright (c) 2016 by TAOS Technologies, Inc. -# All rights reserved. -# -# This file is proprietary and confidential to TAOS Technologies. -# No part of this file may be reproduced, stored, transmitted, -# disclosed or used in any form or by any means other than as -# expressly provided by the written permission from Jianhui Tao -# -################################################################### - -# -*- coding: utf-8 -*- - -import random -import string - -from numpy import logspace -from util import constant -from util.log import * -from util.cases import * -from util.sql import * -from util.common import * -from util.sqlset import TDSetSql - -class TDTestCase: - def init(self, conn, logSql, replicaVar=1): - tdLog.debug("start to execute %s" % __file__) - tdSql.init(conn.cursor(), True) - self.dbname = 'db_test' - self.setsql = TDSetSql() - self.stbname = 'stb' - self.ntbname = 'ntb' - self.rowNum = 5 - self.tbnum = 2 - self.ts = 1537146000000 - self.binary_str = 'taosdata' - self.nchar_str = '涛思数据' - self.str_length = 20 - self.column_dict = { - 'col1': 'tinyint', - 'col2': 'smallint', - 'col3': 'int', - 'col4': 'bigint', - 'col5': 'tinyint unsigned', - 'col6': 'smallint unsigned', - 'col7': 'int unsigned', - 'col8': 'bigint unsigned', - 'col9': 'float', - 'col10': 'double', - 'col11': 'bool', - 'col12': f'binary({self.str_length})', - 'col13': f'nchar({self.str_length})', - - } - - self.tinyint_val = random.randint(constant.TINYINT_MIN,constant.TINYINT_MAX) - self.smallint_val = random.randint(constant.SMALLINT_MIN,constant.SMALLINT_MAX) - self.int_val = random.randint(constant.INT_MIN,constant.INT_MAX) - self.bigint_val = random.randint(constant.BIGINT_MIN,constant.BIGINT_MAX) - self.untingint_val = random.randint(constant.TINYINT_UN_MIN,constant.TINYINT_UN_MAX) - self.unsmallint_val = random.randint(constant.SMALLINT_UN_MIN,constant.SMALLINT_UN_MAX) - self.unint_val = random.randint(constant.INT_UN_MIN,constant.INT_MAX) - self.unbigint_val = random.randint(constant.BIGINT_UN_MIN,constant.BIGINT_UN_MAX) - self.float_val = random.uniform(constant.FLOAT_MIN,constant.FLOAT_MAX) - self.double_val = random.uniform(constant.DOUBLE_MIN*(1E-300),constant.DOUBLE_MAX*(1E-300)) - self.bool_val = random.randint(0,100)%2 - self.binary_val = tdCom.getLongName(random.randint(0,self.str_length)) - self.nchar_val = tdCom.getLongName(random.randint(0,self.str_length)) - self.base_data = { - 'tinyint':self.tinyint_val, - 'smallint':self.smallint_val, - 'int':self.int_val, - 'bigint':self.bigint_val, - 'tinyint unsigned':self.untingint_val, - 'smallint unsigned':self.unsmallint_val, - 'int unsigned':self.unint_val, - 'bigint unsigned':self.unbigint_val, - 'bool':self.bool_val, - 'float':self.float_val, - 'double':self.double_val, - 'binary':self.binary_val, - 'nchar':self.nchar_val - } - - def insert_base_data(self,col_type,tbname,rows,base_data): - for i in range(rows): - if col_type.lower() == 'tinyint': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["tinyint"]})') - elif col_type.lower() == 'smallint': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["smallint"]})') - elif col_type.lower() == 'int': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["int"]})') - elif col_type.lower() == 'bigint': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["bigint"]})') - elif col_type.lower() == 'tinyint unsigned': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["tinyint unsigned"]})') - elif col_type.lower() == 'smallint unsigned': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["smallint unsigned"]})') - elif col_type.lower() == 'int unsigned': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["int unsigned"]})') - elif col_type.lower() == 'bigint unsigned': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["bigint unsigned"]})') - elif col_type.lower() == 'bool': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["bool"]})') - elif col_type.lower() == 'float': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["float"]})') - elif col_type.lower() == 'double': - tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["double"]})') - elif 'binary' in col_type.lower(): - tdSql.execute(f'''insert into {tbname} values({self.ts+i},"{base_data['binary']}")''') - elif 'nchar' in col_type.lower(): - tdSql.execute(f'''insert into {tbname} values({self.ts+i},"{base_data['nchar']}")''') - def delete_all_data(self,tbname,col_type,row_num,base_data,dbname,tb_type,tb_num=1): - tdSql.execute(f'delete from {tbname}') - tdSql.execute(f'flush database {dbname}') - tdSql.execute('reset query cache') - tdSql.query(f'select * from {tbname}') - tdSql.checkRows(0) - if tb_type == 'ntb' or tb_type == 'ctb': - self.insert_base_data(col_type,tbname,row_num,base_data) - elif tb_type == 'stb': - for i in range(tb_num): - self.insert_base_data(col_type,f'{tbname}_{i}',row_num,base_data) - tdSql.execute(f'flush database {dbname}') - tdSql.execute('reset query cache') - tdSql.query(f'select * from {tbname}') - if tb_type == 'ntb' or tb_type == 'ctb': - tdSql.checkRows(row_num) - elif tb_type =='stb': - tdSql.checkRows(row_num*tb_num) - def delete_one_row(self,tbname,column_type,column_name,base_data,row_num,dbname,tb_type,tb_num=1): - tdSql.execute(f'delete from {tbname} where ts={self.ts}') - tdSql.execute(f'flush database {dbname}') - tdSql.execute('reset query cache') - tdSql.query(f'select {column_name} from {tbname}') - if tb_type == 'ntb' or tb_type == 'ctb': - tdSql.checkRows(row_num-1) - elif tb_type == 'stb': - tdSql.checkRows((row_num-1)*tb_num) - tdSql.query(f'select {column_name} from {tbname} where ts={self.ts}') - tdSql.checkRows(0) - if tb_type == 'ntb' or tb_type == 'ctb': - if 'binary' in column_type.lower(): - tdSql.execute(f'''insert into {tbname} values({self.ts},"{base_data['binary']}")''') - elif 'nchar' in column_type.lower(): - tdSql.execute(f'''insert into {tbname} values({self.ts},"{base_data['nchar']}")''') - else: - tdSql.execute(f'insert into {tbname} values({self.ts},{base_data[column_type]})') - elif tb_type == 'stb': - for i in range(tb_num): - if 'binary' in column_type.lower(): - tdSql.execute(f'''insert into {tbname}_{i} values({self.ts},"{base_data['binary']}")''') - elif 'nchar' in column_type.lower(): - tdSql.execute(f'''insert into {tbname}_{i} values({self.ts},"{base_data['nchar']}")''') - else: - tdSql.execute(f'insert into {tbname}_{i} values({self.ts},{base_data[column_type]})') - tdSql.query(f'select {column_name} from {tbname} where ts={self.ts}') - if column_type.lower() == 'float' or column_type.lower() == 'double': - if abs(tdSql.queryResult[0][0] - base_data[column_type]) / base_data[column_type] <= 0.0001: - tdSql.checkEqual(tdSql.queryResult[0][0],tdSql.queryResult[0][0]) - else: - tdLog.exit(f'{column_type} data check failure') - elif 'binary' in column_type.lower(): - tdSql.checkEqual(tdSql.queryResult[0][0],base_data['binary']) - elif 'nchar' in column_type.lower(): - tdSql.checkEqual(tdSql.queryResult[0][0],base_data['nchar']) - else: - tdSql.checkEqual(tdSql.queryResult[0][0],base_data[column_type]) - def delete_rows(self,dbname,tbname,col_name,col_type,base_data,row_num,tb_type,tb_num=1): - for i in range(row_num): - tdSql.execute(f'delete from {tbname} where ts>{self.ts+i}') - tdSql.execute(f'flush database {dbname}') - tdSql.execute('reset query cache') - tdSql.query(f'select {col_name} from {tbname}') - if tb_type == 'ntb' or tb_type == 'ctb': - tdSql.checkRows(i+1) - self.insert_base_data(col_type,tbname,row_num,base_data) - elif tb_type == 'stb': - tdSql.checkRows((i+1)*tb_num) - for j in range(tb_num): - self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) - for i in range(row_num): - tdSql.execute(f'delete from {tbname} where ts>={self.ts+i}') - tdSql.execute(f'flush database {dbname}') - tdSql.execute('reset query cache') - tdSql.query(f'select {col_name} from {tbname}') - if tb_type == 'ntb' or tb_type == 'ctb': - tdSql.checkRows(i) - self.insert_base_data(col_type,tbname,row_num,base_data) - elif tb_type == 'stb': - tdSql.checkRows(i*tb_num) - for j in range(tb_num): - self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) - for i in range(row_num): - tdSql.execute(f'delete from {tbname} where ts<={self.ts+i}') - tdSql.execute(f'flush database {dbname}') - tdSql.execute('reset query cache') - tdSql.query(f'select {col_name} from {tbname}') - if tb_type == 'ntb' or tb_type == 'ctb': - tdSql.checkRows(row_num-i-1) - self.insert_base_data(col_type,tbname,row_num,base_data) - elif tb_type == 'stb': - tdSql.checkRows((row_num-i-1)*tb_num) - for j in range(tb_num): - self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) - for i in range(row_num): - tdSql.execute(f'delete from {tbname} where ts<{self.ts+i}') - tdSql.execute(f'flush database {dbname}') - tdSql.execute('reset query cache') - tdSql.query(f'select {col_name} from {tbname}') - if tb_type == 'ntb' or tb_type == 'ctb': - tdSql.checkRows(row_num-i) - self.insert_base_data(col_type,tbname,row_num,base_data) - elif tb_type == 'stb': - tdSql.checkRows((row_num-i)*tb_num) - for j in range(tb_num): - self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) - for i in range(row_num): - tdSql.execute(f'delete from {tbname} where ts between {self.ts} and {self.ts+i}') - tdSql.execute(f'flush database {dbname}') - tdSql.execute('reset query cache') - tdSql.query(f'select {col_name} from {tbname}') - if tb_type == 'ntb' or tb_type == 'ctb': - tdSql.checkRows(row_num - i-1) - self.insert_base_data(col_type,tbname,row_num,base_data) - elif tb_type == 'stb': - tdSql.checkRows(tb_num*(row_num - i-1)) - for j in range(tb_num): - self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) - tdSql.execute(f'delete from {tbname} where ts between {self.ts+i+1} and {self.ts}') - tdSql.query(f'select {col_name} from {tbname}') - if tb_type == 'ntb' or tb_type == 'ctb': - tdSql.checkRows(row_num) - elif tb_type == 'stb': - tdSql.checkRows(tb_num*row_num) - def delete_error(self,tbname,column_name,column_type,base_data): - for error_list in ['',f'ts = {self.ts} and',f'ts = {self.ts} or']: - if 'binary' in column_type.lower(): - tdSql.error(f'''delete from {tbname} where {error_list} {column_name} ="{base_data['binary']}"''') - elif 'nchar' in column_type.lower(): - tdSql.error(f'''delete from {tbname} where {error_list} {column_name} ="{base_data['nchar']}"''') - else: - tdSql.error(f'delete from {tbname} where {error_list} {column_name} = {base_data[column_type]}') - - def delete_data_ntb(self): - tdSql.execute(f'create database if not exists {self.dbname}') - tdSql.execute(f'use {self.dbname}') - for col_name,col_type in self.column_dict.items(): - tdSql.execute(f'create table {self.ntbname} (ts timestamp,{col_name} {col_type})') - self.insert_base_data(col_type,self.ntbname,self.rowNum,self.base_data) - self.delete_one_row(self.ntbname,col_type,col_name,self.base_data,self.rowNum,self.dbname,'ntb') - self.delete_all_data(self.ntbname,col_type,self.rowNum,self.base_data,self.dbname,'ntb') - self.delete_error(self.ntbname,col_name,col_type,self.base_data) - self.delete_rows(self.dbname,self.ntbname,col_name,col_type,self.base_data,self.rowNum,'ntb') - for func in ['first','last']: - tdSql.query(f'select {func}(*) from {self.ntbname}') - tdSql.execute(f'drop table {self.ntbname}') - tdSql.execute(f'drop database {self.dbname}') - def delete_data_ctb(self): - tdSql.execute(f'create database if not exists {self.dbname}') - tdSql.execute(f'use {self.dbname}') - for col_name,col_type in self.column_dict.items(): - tdSql.execute(f'create table {self.stbname} (ts timestamp,{col_name} {col_type}) tags(t1 int)') - for i in range(self.tbnum): - tdSql.execute(f'create table {self.stbname}_{i} using {self.stbname} tags(1)') - self.insert_base_data(col_type,f'{self.stbname}_{i}',self.rowNum,self.base_data) - self.delete_one_row(f'{self.stbname}_{i}',col_type,col_name,self.base_data,self.rowNum,self.dbname,'ctb') - self.delete_all_data(f'{self.stbname}_{i}',col_type,self.rowNum,self.base_data,self.dbname,'ctb') - self.delete_error(f'{self.stbname}_{i}',col_name,col_type,self.base_data) - self.delete_rows(self.dbname,f'{self.stbname}_{i}',col_name,col_type,self.base_data,self.rowNum,'ctb') - for func in ['first','last']: - tdSql.query(f'select {func}(*) from {self.stbname}_{i}') - tdSql.execute(f'drop table {self.stbname}') - def delete_data_stb(self): - tdSql.execute(f'create database if not exists {self.dbname}') - tdSql.execute(f'use {self.dbname}') - for col_name,col_type in self.column_dict.items(): - tdSql.execute(f'create table {self.stbname} (ts timestamp,{col_name} {col_type}) tags(t1 int)') - for i in range(self.tbnum): - tdSql.execute(f'create table {self.stbname}_{i} using {self.stbname} tags(1)') - self.insert_base_data(col_type,f'{self.stbname}_{i}',self.rowNum,self.base_data) - self.delete_error(self.stbname,col_name,col_type,self.base_data) - self.delete_one_row(self.stbname,col_type,col_name,self.base_data,self.rowNum,self.dbname,'stb',self.tbnum) - self.delete_all_data(self.stbname,col_type,self.rowNum,self.base_data,self.dbname,'stb',self.tbnum) - self.delete_rows(self.dbname,self.stbname,col_name,col_type,self.base_data,self.rowNum,'stb',self.tbnum) - for func in ['first','last']: - tdSql.query(f'select {func}(*) from {self.stbname}') - tdSql.execute(f'drop table {self.stbname}') - tdSql.execute(f'drop database {self.dbname}') - def run(self): - self.delete_data_ntb() - self.delete_data_ctb() - self.delete_data_stb() - tdDnodes.stoptaosd(1) - tdDnodes.starttaosd(1) - self.delete_data_ntb() - def stop(self): - tdSql.close() - tdLog.success("%s successfully executed" % __file__) - -tdCases.addWindows(__file__, TDTestCase()) -tdCases.addLinux(__file__, TDTestCase()) +################################################################### +# Copyright (c) 2016 by TAOS Technologies, Inc. +# All rights reserved. +# +# This file is proprietary and confidential to TAOS Technologies. +# No part of this file may be reproduced, stored, transmitted, +# disclosed or used in any form or by any means other than as +# expressly provided by the written permission from Jianhui Tao +# +################################################################### + +# -*- coding: utf-8 -*- + +import random +import string + +from numpy import logspace +from util import constant +from util.log import * +from util.cases import * +from util.sql import * +from util.common import * +from util.sqlset import TDSetSql + +class TDTestCase: + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor(), True) + self.dbname = 'db_test' + self.setsql = TDSetSql() + self.stbname = 'stb' + self.ntbname = 'ntb' + self.rowNum = 10 + self.tbnum = 3 + self.ts = 1537146000000 + self.binary_str = 'taosdata' + self.nchar_str = '涛思数据' + self.str_length = 20 + self.column_dict = { + 'col1': 'tinyint', + 'col2': 'smallint', + 'col3': 'int', + 'col4': 'bigint', + 'col5': 'tinyint unsigned', + 'col6': 'smallint unsigned', + 'col7': 'int unsigned', + 'col8': 'bigint unsigned', + 'col9': 'float', + 'col10': 'double', + 'col11': 'bool', + 'col12': f'binary({self.str_length})', + 'col13': f'nchar({self.str_length})', + + } + + self.tinyint_val = random.randint(constant.TINYINT_MIN,constant.TINYINT_MAX) + self.smallint_val = random.randint(constant.SMALLINT_MIN,constant.SMALLINT_MAX) + self.int_val = random.randint(constant.INT_MIN,constant.INT_MAX) + self.bigint_val = random.randint(constant.BIGINT_MIN,constant.BIGINT_MAX) + self.untingint_val = random.randint(constant.TINYINT_UN_MIN,constant.TINYINT_UN_MAX) + self.unsmallint_val = random.randint(constant.SMALLINT_UN_MIN,constant.SMALLINT_UN_MAX) + self.unint_val = random.randint(constant.INT_UN_MIN,constant.INT_MAX) + self.unbigint_val = random.randint(constant.BIGINT_UN_MIN,constant.BIGINT_UN_MAX) + self.float_val = random.uniform(constant.FLOAT_MIN,constant.FLOAT_MAX) + self.double_val = random.uniform(constant.DOUBLE_MIN*(1E-300),constant.DOUBLE_MAX*(1E-300)) + self.bool_val = random.randint(0,100)%2 + self.binary_val = tdCom.getLongName(random.randint(0,self.str_length)) + self.nchar_val = tdCom.getLongName(random.randint(0,self.str_length)) + self.base_data = { + 'tinyint':self.tinyint_val, + 'smallint':self.smallint_val, + 'int':self.int_val, + 'bigint':self.bigint_val, + 'tinyint unsigned':self.untingint_val, + 'smallint unsigned':self.unsmallint_val, + 'int unsigned':self.unint_val, + 'bigint unsigned':self.unbigint_val, + 'bool':self.bool_val, + 'float':self.float_val, + 'double':self.double_val, + 'binary':self.binary_val, + 'nchar':self.nchar_val + } + + def insert_base_data(self,col_type,tbname,rows,base_data): + for i in range(rows): + if col_type.lower() == 'tinyint': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["tinyint"]})') + elif col_type.lower() == 'smallint': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["smallint"]})') + elif col_type.lower() == 'int': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["int"]})') + elif col_type.lower() == 'bigint': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["bigint"]})') + elif col_type.lower() == 'tinyint unsigned': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["tinyint unsigned"]})') + elif col_type.lower() == 'smallint unsigned': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["smallint unsigned"]})') + elif col_type.lower() == 'int unsigned': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["int unsigned"]})') + elif col_type.lower() == 'bigint unsigned': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["bigint unsigned"]})') + elif col_type.lower() == 'bool': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["bool"]})') + elif col_type.lower() == 'float': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["float"]})') + elif col_type.lower() == 'double': + tdSql.execute(f'insert into {tbname} values({self.ts+i},{base_data["double"]})') + elif 'binary' in col_type.lower(): + tdSql.execute(f'''insert into {tbname} values({self.ts+i},"{base_data['binary']}")''') + elif 'nchar' in col_type.lower(): + tdSql.execute(f'''insert into {tbname} values({self.ts+i},"{base_data['nchar']}")''') + def delete_all_data(self,tbname,col_type,row_num,base_data,dbname,tb_type,tb_num=1,stbname=''): + tdSql.query(f'select count(*) from {tbname}') + tdSql.execute(f'delete from {tbname}') + tdSql.execute(f'flush database {dbname}') + tdSql.execute('reset query cache') + tdSql.query(f'select * from {tbname}') + tdSql.checkRows(0) + if tb_type == 'ntb' or tb_type == 'ctb': + if tb_type == 'ctb': + tdSql.query(f'select count(*) from {stbname}') + if tb_num <= 1: + if len(tdSql.queryResult) != 0: + tdLog.exit('delete case failure!') + else: + tdSql.checkEqual(tdSql.queryResult[0][0],(tb_num-1)*row_num) + + self.insert_base_data(col_type,tbname,row_num,base_data) + elif tb_type == 'stb': + for i in range(tb_num): + self.insert_base_data(col_type,f'{tbname}_{i}',row_num,base_data) + tdSql.execute(f'flush database {dbname}') + tdSql.execute('reset query cache') + tdSql.query(f'select * from {tbname}') + if tb_type == 'ntb' or tb_type == 'ctb': + tdSql.checkRows(row_num) + elif tb_type =='stb': + tdSql.checkRows(row_num*tb_num) + def delete_one_row(self,tbname,column_type,column_name,base_data,row_num,dbname,tb_type,tb_num=1): + tdSql.execute(f'delete from {tbname} where ts={self.ts}') + tdSql.execute(f'flush database {dbname}') + tdSql.execute('reset query cache') + tdSql.query(f'select {column_name} from {tbname}') + if tb_type == 'ntb' or tb_type == 'ctb': + tdSql.checkRows(row_num-1) + elif tb_type == 'stb': + tdSql.checkRows((row_num-1)*tb_num) + tdSql.query(f'select {column_name} from {tbname} where ts={self.ts}') + tdSql.checkRows(0) + if tb_type == 'ntb' or tb_type == 'ctb': + if 'binary' in column_type.lower(): + tdSql.execute(f'''insert into {tbname} values({self.ts},"{base_data['binary']}")''') + elif 'nchar' in column_type.lower(): + tdSql.execute(f'''insert into {tbname} values({self.ts},"{base_data['nchar']}")''') + else: + tdSql.execute(f'insert into {tbname} values({self.ts},{base_data[column_type]})') + elif tb_type == 'stb': + for i in range(tb_num): + if 'binary' in column_type.lower(): + tdSql.execute(f'''insert into {tbname}_{i} values({self.ts},"{base_data['binary']}")''') + elif 'nchar' in column_type.lower(): + tdSql.execute(f'''insert into {tbname}_{i} values({self.ts},"{base_data['nchar']}")''') + else: + tdSql.execute(f'insert into {tbname}_{i} values({self.ts},{base_data[column_type]})') + tdSql.query(f'select {column_name} from {tbname} where ts={self.ts}') + if column_type.lower() == 'float' or column_type.lower() == 'double': + if abs(tdSql.queryResult[0][0] - base_data[column_type]) / base_data[column_type] <= 0.0001: + tdSql.checkEqual(tdSql.queryResult[0][0],tdSql.queryResult[0][0]) + else: + tdLog.exit(f'{column_type} data check failure') + elif 'binary' in column_type.lower(): + tdSql.checkEqual(tdSql.queryResult[0][0],base_data['binary']) + elif 'nchar' in column_type.lower(): + tdSql.checkEqual(tdSql.queryResult[0][0],base_data['nchar']) + else: + tdSql.checkEqual(tdSql.queryResult[0][0],base_data[column_type]) + def delete_rows(self,dbname,tbname,col_name,col_type,base_data,row_num,tb_type,tb_num=1): + for i in range(row_num): + tdSql.execute(f'delete from {tbname} where ts>{self.ts+i}') + tdSql.execute(f'flush database {dbname}') + tdSql.execute('reset query cache') + tdSql.query(f'select {col_name} from {tbname}') + if tb_type == 'ntb' or tb_type == 'ctb': + tdSql.checkRows(i+1) + self.insert_base_data(col_type,tbname,row_num,base_data) + elif tb_type == 'stb': + tdSql.checkRows((i+1)*tb_num) + for j in range(tb_num): + self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) + for i in range(row_num): + tdSql.execute(f'delete from {tbname} where ts>={self.ts+i}') + tdSql.execute(f'flush database {dbname}') + tdSql.execute('reset query cache') + tdSql.query(f'select {col_name} from {tbname}') + if tb_type == 'ntb' or tb_type == 'ctb': + tdSql.checkRows(i) + self.insert_base_data(col_type,tbname,row_num,base_data) + elif tb_type == 'stb': + tdSql.checkRows(i*tb_num) + for j in range(tb_num): + self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) + for i in range(row_num): + tdSql.execute(f'delete from {tbname} where ts<={self.ts+i}') + tdSql.execute(f'flush database {dbname}') + tdSql.execute('reset query cache') + tdSql.query(f'select {col_name} from {tbname}') + if tb_type == 'ntb' or tb_type == 'ctb': + tdSql.checkRows(row_num-i-1) + self.insert_base_data(col_type,tbname,row_num,base_data) + elif tb_type == 'stb': + tdSql.checkRows((row_num-i-1)*tb_num) + for j in range(tb_num): + self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) + for i in range(row_num): + tdSql.execute(f'delete from {tbname} where ts<{self.ts+i}') + tdSql.execute(f'flush database {dbname}') + tdSql.execute('reset query cache') + tdSql.query(f'select {col_name} from {tbname}') + if tb_type == 'ntb' or tb_type == 'ctb': + tdSql.checkRows(row_num-i) + self.insert_base_data(col_type,tbname,row_num,base_data) + elif tb_type == 'stb': + tdSql.checkRows((row_num-i)*tb_num) + for j in range(tb_num): + self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) + for i in range(row_num): + tdSql.execute(f'delete from {tbname} where ts between {self.ts} and {self.ts+i}') + tdSql.execute(f'flush database {dbname}') + tdSql.execute('reset query cache') + tdSql.query(f'select {col_name} from {tbname}') + if tb_type == 'ntb' or tb_type == 'ctb': + tdSql.checkRows(row_num - i-1) + self.insert_base_data(col_type,tbname,row_num,base_data) + elif tb_type == 'stb': + tdSql.checkRows(tb_num*(row_num - i-1)) + for j in range(tb_num): + self.insert_base_data(col_type,f'{tbname}_{j}',row_num,base_data) + tdSql.execute(f'delete from {tbname} where ts between {self.ts+i+1} and {self.ts}') + tdSql.query(f'select {col_name} from {tbname}') + if tb_type == 'ntb' or tb_type == 'ctb': + tdSql.checkRows(row_num) + elif tb_type == 'stb': + tdSql.checkRows(tb_num*row_num) + def delete_error(self,tbname,column_name,column_type,base_data): + for error_list in ['',f'ts = {self.ts} and',f'ts = {self.ts} or']: + if 'binary' in column_type.lower(): + tdSql.error(f'''delete from {tbname} where {error_list} {column_name} ="{base_data['binary']}"''') + elif 'nchar' in column_type.lower(): + tdSql.error(f'''delete from {tbname} where {error_list} {column_name} ="{base_data['nchar']}"''') + else: + tdSql.error(f'delete from {tbname} where {error_list} {column_name} = {base_data[column_type]}') + + def delete_data_ntb(self): + tdSql.execute(f'create database if not exists {self.dbname}') + tdSql.execute(f'use {self.dbname}') + for col_name,col_type in self.column_dict.items(): + tdSql.execute(f'create table {self.ntbname} (ts timestamp,{col_name} {col_type})') + self.insert_base_data(col_type,self.ntbname,self.rowNum,self.base_data) + self.delete_one_row(self.ntbname,col_type,col_name,self.base_data,self.rowNum,self.dbname,'ntb') + self.delete_all_data(self.ntbname,col_type,self.rowNum,self.base_data,self.dbname,'ntb') + self.delete_error(self.ntbname,col_name,col_type,self.base_data) + self.delete_rows(self.dbname,self.ntbname,col_name,col_type,self.base_data,self.rowNum,'ntb') + for func in ['first','last']: + tdSql.query(f'select {func}(*) from {self.ntbname}') + tdSql.execute(f'drop table {self.ntbname}') + tdSql.execute(f'drop database {self.dbname}') + def delete_data_ctb(self): + tdSql.execute(f'create database if not exists {self.dbname}') + tdSql.execute(f'use {self.dbname}') + for col_name,col_type in self.column_dict.items(): + tdSql.execute(f'create table {self.stbname} (ts timestamp,{col_name} {col_type}) tags(t1 int)') + for i in range(self.tbnum): + tdSql.execute(f'create table {self.stbname}_{i} using {self.stbname} tags(1)') + self.insert_base_data(col_type,f'{self.stbname}_{i}',self.rowNum,self.base_data) + self.delete_one_row(f'{self.stbname}_{i}',col_type,col_name,self.base_data,self.rowNum,self.dbname,'ctb') + self.delete_all_data(f'{self.stbname}_{i}',col_type,self.rowNum,self.base_data,self.dbname,'ctb',i+1,self.stbname) + self.delete_error(f'{self.stbname}_{i}',col_name,col_type,self.base_data) + self.delete_rows(self.dbname,f'{self.stbname}_{i}',col_name,col_type,self.base_data,self.rowNum,'ctb') + for func in ['first','last']: + tdSql.query(f'select {func}(*) from {self.stbname}_{i}') + tdSql.execute(f'drop table {self.stbname}') + def delete_data_stb(self): + tdSql.execute(f'create database if not exists {self.dbname}') + tdSql.execute(f'use {self.dbname}') + for col_name,col_type in self.column_dict.items(): + tdSql.execute(f'create table {self.stbname} (ts timestamp,{col_name} {col_type}) tags(t1 int)') + for i in range(self.tbnum): + tdSql.execute(f'create table {self.stbname}_{i} using {self.stbname} tags(1)') + self.insert_base_data(col_type,f'{self.stbname}_{i}',self.rowNum,self.base_data) + self.delete_error(self.stbname,col_name,col_type,self.base_data) + self.delete_one_row(self.stbname,col_type,col_name,self.base_data,self.rowNum,self.dbname,'stb',self.tbnum) + self.delete_all_data(self.stbname,col_type,self.rowNum,self.base_data,self.dbname,'stb',self.tbnum) + self.delete_rows(self.dbname,self.stbname,col_name,col_type,self.base_data,self.rowNum,'stb',self.tbnum) + for func in ['first','last']: + tdSql.query(f'select {func}(*) from {self.stbname}') + tdSql.execute(f'drop table {self.stbname}') + tdSql.execute(f'drop database {self.dbname}') + def run(self): + self.delete_data_ntb() + self.delete_data_ctb() + self.delete_data_stb() + tdDnodes.stoptaosd(1) + tdDnodes.starttaosd(1) + self.delete_data_ntb() + def stop(self): + tdSql.close() + tdLog.success("%s successfully executed" % __file__) + +tdCases.addWindows(__file__, TDTestCase()) +tdCases.addLinux(__file__, TDTestCase())