diff --git a/docs/examples/java/src/main/java/com/taos/example/StmtInsertExample.java b/docs/examples/java/src/main/java/com/taos/example/StmtInsertExample.java index bbcc92b22f67c31384b0fb7a082975eaac2ff2bc..72d4ecc725aa4c7cfd6543dd0c41d70563252c1d 100644 --- a/docs/examples/java/src/main/java/com/taos/example/StmtInsertExample.java +++ b/docs/examples/java/src/main/java/com/taos/example/StmtInsertExample.java @@ -6,39 +6,32 @@ import java.sql.Connection; import java.sql.DriverManager; import java.sql.SQLException; import java.sql.Statement; +import java.text.SimpleDateFormat; import java.time.LocalDateTime; import java.time.ZoneOffset; import java.time.format.DateTimeFormatter; import java.util.ArrayList; -import java.util.Arrays; +import java.util.Comparator; import java.util.List; +import java.util.Random; +import java.util.stream.Collectors; public class StmtInsertExample { - private static ArrayList tsToLongArray(String ts) { - ArrayList result = new ArrayList<>(); - DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS"); - LocalDateTime localDateTime = LocalDateTime.parse(ts, formatter); - result.add(localDateTime.toInstant(ZoneOffset.of("+8")).toEpochMilli()); - return result; - } - - private static ArrayList toArray(T v) { - ArrayList result = new ArrayList<>(); - result.add(v); - return result; - } + private static String datePattern = "yyyy-MM-dd HH:mm:ss.SSS"; + private static DateTimeFormatter formatter = DateTimeFormatter.ofPattern(datePattern); - private static List getRawData() { - return Arrays.asList( - "d1001,2018-10-03 14:38:05.000,10.30000,219,0.31000,California.SanFrancisco,2", - "d1001,2018-10-03 14:38:15.000,12.60000,218,0.33000,California.SanFrancisco,2", - "d1001,2018-10-03 14:38:16.800,12.30000,221,0.31000,California.SanFrancisco,2", - "d1002,2018-10-03 14:38:16.650,10.30000,218,0.25000,California.SanFrancisco,3", - "d1003,2018-10-03 14:38:05.500,11.80000,221,0.28000,California.LosAngeles,2", - "d1003,2018-10-03 14:38:16.600,13.40000,223,0.29000,California.LosAngeles,2", - "d1004,2018-10-03 14:38:05.000,10.80000,223,0.29000,California.LosAngeles,3", - "d1004,2018-10-03 14:38:06.500,11.50000,221,0.35000,California.LosAngeles,3" - ); + private static List getRawData(int size) { + SimpleDateFormat format = new SimpleDateFormat(datePattern); + List result = new ArrayList<>(); + long current = System.currentTimeMillis(); + Random random = new Random(); + for (int i = 0; i < size; i++) { + String time = format.format(current + i); + int id = random.nextInt(10); + result.add("d" + id + "," + time + ",10.30000,219,0.31000,California.SanFrancisco,2"); + } + return result.stream() + .sorted(Comparator.comparing(s -> s.split(",")[0])).collect(Collectors.toList()); } private static Connection getConnection() throws SQLException { @@ -48,9 +41,9 @@ public class StmtInsertExample { private static void createTable(Connection conn) throws SQLException { try (Statement stmt = conn.createStatement()) { - stmt.execute("CREATE DATABASE power KEEP 3650"); - stmt.executeUpdate("USE power"); - stmt.execute("CREATE STABLE meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) " + + stmt.execute("CREATE DATABASE if not exists power KEEP 3650"); + stmt.executeUpdate("use power"); + stmt.execute("CREATE STABLE if not exists meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) " + "TAGS (location BINARY(64), groupId INT)"); } } @@ -58,21 +51,54 @@ public class StmtInsertExample { private static void insertData() throws SQLException { try (Connection conn = getConnection()) { createTable(conn); - String psql = "INSERT INTO ? USING meters TAGS(?, ?) VALUES(?, ?, ?, ?)"; + String psql = "INSERT INTO ? USING power.meters TAGS(?, ?) VALUES(?, ?, ?, ?)"; try (TSDBPreparedStatement pst = (TSDBPreparedStatement) conn.prepareStatement(psql)) { - for (String line : getRawData()) { + String tableName = null; + ArrayList ts = new ArrayList<>(); + ArrayList current = new ArrayList<>(); + ArrayList voltage = new ArrayList<>(); + ArrayList phase = new ArrayList<>(); + for (String line : getRawData(100000)) { String[] ps = line.split(","); - // bind table name and tags - pst.setTableName(ps[0]); - pst.setTagString(0, ps[5]); - pst.setTagInt(1, Integer.valueOf(ps[6])); + if (tableName == null) { + // bind table name and tags + tableName = "power." + ps[0]; + pst.setTableName(ps[0]); + pst.setTagString(0, ps[5]); + pst.setTagInt(1, Integer.valueOf(ps[6])); + } else { + if (!tableName.equals(ps[0])) { + pst.setTimestamp(0, ts); + pst.setFloat(1, current); + pst.setInt(2, voltage); + pst.setFloat(3, phase); + pst.columnDataAddBatch(); + pst.columnDataExecuteBatch(); + + // bind table name and tags + tableName = ps[0]; + pst.setTableName(ps[0]); + pst.setTagString(0, ps[5]); + pst.setTagInt(1, Integer.valueOf(ps[6])); + ts.clear(); + current.clear(); + voltage.clear(); + phase.clear(); + } + } // bind values - pst.setTimestamp(0, tsToLongArray(ps[1])); //ps[1] looks like: 2018-10-03 14:38:05.000 - pst.setFloat(1, toArray(Float.valueOf(ps[2]))); - pst.setInt(2, toArray(Integer.valueOf(ps[3]))); - pst.setFloat(3, toArray(Float.valueOf(ps[4]))); - pst.columnDataAddBatch(); + // ps[1] looks like: 2018-10-03 14:38:05.000 + LocalDateTime localDateTime = LocalDateTime.parse(ps[1], formatter); + ts.add(localDateTime.toInstant(ZoneOffset.of("+8")).toEpochMilli()); + current.add(Float.valueOf(ps[2])); + voltage.add(Integer.valueOf(ps[3])); + phase.add(Float.valueOf(ps[4])); } + pst.setTimestamp(0, ts); + pst.setFloat(1, current); + pst.setInt(2, voltage); + pst.setFloat(3, phase); + pst.columnDataAddBatch(); pst.columnDataExecuteBatch(); } } diff --git a/packaging/tools/makepkg.sh b/packaging/tools/makepkg.sh index 45ef20a22d0fd65143afb4c07b7c70f16a362eb4..2f1e803689cb877560ca7d41bcd3a16ff1264b05 100755 --- a/packaging/tools/makepkg.sh +++ b/packaging/tools/makepkg.sh @@ -429,7 +429,7 @@ if [ "$exitcode" != "0" ]; then exit $exitcode fi -if [ -n "${taostools_bin_files}" ]; then +if [ -n "${taostools_bin_files}" ] && [ "$verMode" != "cloud" ]; then wget https://github.com/taosdata/grafanaplugin/releases/latest/download/TDinsight.sh -O ${taostools_install_dir}/bin/TDinsight.sh && echo "TDinsight.sh downloaded!"|| echo "failed to download TDinsight.sh" if [ "$osType" != "Darwin" ]; then tar -zcv -f "$(basename ${taostools_pkg_name}).tar.gz" "$(basename ${taostools_install_dir})" --remove-files || : diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 4a872482b4c1bf447c4d977bfd0977a200a22f85..311c79381cc8c7919beaea26e120c3ea08e8a11f 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -1970,7 +1970,7 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf) if (len >= size - 1) return dumpBuf; for (int32_t j = 0; j < rows; j++) { - len += snprintf(dumpBuf + len, size - len, "%s %d|", flag, j); + len += snprintf(dumpBuf + len, size - len, "%s|", flag); if (len >= size - 1) return dumpBuf; for (int32_t k = 0; k < colNum; k++) { @@ -2053,7 +2053,7 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf) } break; } } - len += snprintf(dumpBuf + len, size - len, "\n"); + len += snprintf(dumpBuf + len, size - len, "%d\n", j); if (len >= size - 1) return dumpBuf; } len += snprintf(dumpBuf + len, size - len, "%s |end\n", flag); diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c index 71c74997bc22658656ead758127166f97c9e4dbc..d33f04c1454b414caadacc7daa7feede0b9410cd 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCache.c +++ b/source/dnode/vnode/src/tsdb/tsdbCache.c @@ -404,16 +404,47 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow for (int i = 0; i < num_keys; ++i) { SColVal *pColVal = (SColVal *)taosArrayGet(aColVal, i); - if (!COL_VAL_IS_NONE(pColVal)) { - SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i + num_keys]); + // if (!COL_VAL_IS_NONE(pColVal)) { + SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i + num_keys]); + + if (NULL == pLastCol || pLastCol->ts <= keyTs) { + char *value = NULL; + size_t vlen = 0; + tsdbCacheSerialize(&(SLastCol){.ts = keyTs, .colVal = *pColVal}, &value, &vlen); + SLastKey key = (SLastKey){.ltype = 0, .uid = uid, .cid = pColVal->cid}; + size_t klen = ROCKS_KEY_LEN; + rocksdb_writebatch_put(wb, (char *)&key, klen, value, vlen); + + pLastCol = (SLastCol *)value; + SLastCol *pTmpLastCol = taosMemoryCalloc(1, sizeof(SLastCol)); + *pTmpLastCol = *pLastCol; + pLastCol = pTmpLastCol; + + reallocVarData(&pLastCol->colVal); + size_t charge = sizeof(*pLastCol); + if (IS_VAR_DATA_TYPE(pLastCol->colVal.type)) { + charge += pLastCol->colVal.value.nData; + } + + LRUStatus status = taosLRUCacheInsert(pTsdb->lruCache, &key, ROCKS_KEY_LEN, pLastCol, charge, tsdbCacheDeleter, + NULL, TAOS_LRU_PRIORITY_LOW); + if (status != TAOS_LRU_STATUS_OK) { + code = -1; + } + + taosMemoryFree(value); + } + + if (COL_VAL_IS_VALUE(pColVal)) { + SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i]); if (NULL == pLastCol || pLastCol->ts <= keyTs) { char *value = NULL; size_t vlen = 0; tsdbCacheSerialize(&(SLastCol){.ts = keyTs, .colVal = *pColVal}, &value, &vlen); - SLastKey key = (SLastKey){.ltype = 0, .uid = uid, .cid = pColVal->cid}; - size_t klen = ROCKS_KEY_LEN; - rocksdb_writebatch_put(wb, (char *)&key, klen, value, vlen); + SLastKey key = (SLastKey){.ltype = 1, .uid = uid, .cid = pColVal->cid}; + + rocksdb_writebatch_put(wb, (char *)&key, ROCKS_KEY_LEN, value, vlen); pLastCol = (SLastCol *)value; SLastCol *pTmpLastCol = taosMemoryCalloc(1, sizeof(SLastCol)); @@ -434,39 +465,8 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow taosMemoryFree(value); } - - if (COL_VAL_IS_VALUE(pColVal)) { - SLastCol *pLastCol = tsdbCacheDeserialize(values_list[i]); - - if (NULL == pLastCol || pLastCol->ts <= keyTs) { - char *value = NULL; - size_t vlen = 0; - tsdbCacheSerialize(&(SLastCol){.ts = keyTs, .colVal = *pColVal}, &value, &vlen); - SLastKey key = (SLastKey){.ltype = 1, .uid = uid, .cid = pColVal->cid}; - - rocksdb_writebatch_put(wb, (char *)&key, ROCKS_KEY_LEN, value, vlen); - - pLastCol = (SLastCol *)value; - SLastCol *pTmpLastCol = taosMemoryCalloc(1, sizeof(SLastCol)); - *pTmpLastCol = *pLastCol; - pLastCol = pTmpLastCol; - - reallocVarData(&pLastCol->colVal); - size_t charge = sizeof(*pLastCol); - if (IS_VAR_DATA_TYPE(pLastCol->colVal.type)) { - charge += pLastCol->colVal.value.nData; - } - - LRUStatus status = taosLRUCacheInsert(pTsdb->lruCache, &key, ROCKS_KEY_LEN, pLastCol, charge, - tsdbCacheDeleter, NULL, TAOS_LRU_PRIORITY_LOW); - if (status != TAOS_LRU_STATUS_OK) { - code = -1; - } - - taosMemoryFree(value); - } - } } + //} rocksdb_free(values_list[i]); rocksdb_free(values_list[i + num_keys]); @@ -474,7 +474,7 @@ int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSDBROW *pRow taosMemoryFree(values_list); taosMemoryFree(values_list_sizes); - rocksMayWrite(pTsdb, false, false, false); + rocksMayWrite(pTsdb, true, false, false); taosThreadMutexUnlock(&pTsdb->rCache.rMutex); _exit: @@ -3010,17 +3010,17 @@ static int32_t mergeLastRowCid(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, memcpy(pCol->colVal.value.pData, pColVal->value.pData, pColVal->value.nData); } - if (COL_VAL_IS_NONE(pColVal)) { + /*if (COL_VAL_IS_NONE(pColVal)) { if (!setNoneCol) { noneCol = iCol; setNoneCol = true; } - } else { - int32_t aColIndex = taosArraySearchIdx(aColArray, &pColVal->cid, compareInt16Val, TD_EQ); - if (aColIndex >= 0) { - taosArrayRemove(aColArray, aColIndex); - } + } else {*/ + int32_t aColIndex = taosArraySearchIdx(aColArray, &pColVal->cid, compareInt16Val, TD_EQ); + if (aColIndex >= 0) { + taosArrayRemove(aColArray, aColIndex); } + //} } if (!setNoneCol) { // done, goto return pColArray diff --git a/source/dnode/vnode/src/tsdb/tsdbCacheRead.c b/source/dnode/vnode/src/tsdb/tsdbCacheRead.c index b93294e46f7f51f50010d1b6fd37a3a8a16ec6bb..0e0b7a2ffa8105774c2f3ffb61eb49fdaf3dd220 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCacheRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbCacheRead.c @@ -343,6 +343,14 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32 hasRes = true; p->ts = pColVal->ts; + if (k == 0) { + if (TARRAY_SIZE(pTableUidList) == 0) { + taosArrayPush(pTableUidList, &pKeyInfo->uid); + } else { + taosArraySet(pTableUidList, 0, &pKeyInfo->uid); + } + } + if (pColVal->ts < singleTableLastTs && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) { singleTableLastTs = pColVal->ts; } @@ -373,12 +381,6 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32 } } - if (TARRAY_SIZE(pTableUidList) == 0) { - taosArrayPush(pTableUidList, &pKeyInfo->uid); - } else { - taosArraySet(pTableUidList, 0, &pKeyInfo->uid); - } - // taosArrayClearEx(pRow, freeItem); taosArrayClear(pRow); } diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 903975a6187ff8a089af277f30595503722447fa..9d882a90ce8300feca915b5900a247cfad4c2bec 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -448,15 +448,17 @@ static bool setTimeWindowInterpolationEndTs(SIntervalAggOperatorInfo* pInfo, SEx return true; } -bool inCalSlidingWindow(SInterval* pInterval, STimeWindow* pWin, TSKEY calStart, TSKEY calEnd) { - if (pInterval->interval != pInterval->sliding && (pWin->ekey < calStart || pWin->skey > calEnd)) { +bool inCalSlidingWindow(SInterval* pInterval, STimeWindow* pWin, TSKEY calStart, TSKEY calEnd, EStreamType blockType) { + if (pInterval->interval != pInterval->sliding && + ((pWin->ekey < calStart || pWin->skey > calEnd) || (blockType == STREAM_PULL_DATA && pWin->skey < calStart) )) { return false; } + return true; } bool inSlidingWindow(SInterval* pInterval, STimeWindow* pWin, SDataBlockInfo* pBlockInfo) { - return inCalSlidingWindow(pInterval, pWin, pBlockInfo->calWin.skey, pBlockInfo->calWin.ekey); + return inCalSlidingWindow(pInterval, pWin, pBlockInfo->calWin.skey, pBlockInfo->calWin.ekey, pBlockInfo->type); } static int32_t getNextQualifiedWindow(SInterval* pInterval, STimeWindow* pNext, SDataBlockInfo* pDataBlockInfo, @@ -1365,7 +1367,7 @@ static void doDeleteWindows(SOperatorInfo* pOperator, SInterval* pInterval, SSDa } do { - if (!inCalSlidingWindow(pInterval, &win, calStTsCols[i], calEnTsCols[i])) { + if (!inCalSlidingWindow(pInterval, &win, calStTsCols[i], calEnTsCols[i], pBlock->info.type)) { getNextTimeWindow(pInterval, pInterval->precision, TSDB_ORDER_ASC, &win); continue; } diff --git a/tests/pytest/util/dnodes.py b/tests/pytest/util/dnodes.py index 5948cde81d5574a1f872b64006537042677a16f4..543433b4ea6a672f10fb341b113d1031f07a1fb2 100644 --- a/tests/pytest/util/dnodes.py +++ b/tests/pytest/util/dnodes.py @@ -130,6 +130,18 @@ class TDDnode: "locale": "en_US.UTF-8", "charset": "UTF-8", "asyncLog": "0", + "mDebugFlag": "143", + "dDebugFlag": "143", + "vDebugFlag": "143", + "tqDebugFlag": "143", + "cDebugFlag": "143", + "jniDebugFlag": "143", + "qDebugFlag": "143", + "rpcDebugFlag": "143", + "tmrDebugFlag": "131", + "uDebugFlag": "143", + "sDebugFlag": "143", + "wDebugFlag": "143", "numOfLogLines": "100000000", "statusInterval": "1", "enableQueryHb": "1", diff --git a/tests/system-test/forcedrop b/tests/system-test/forcedrop deleted file mode 120000 index d2a16bf505dcf2c6d38206b08a9b6af2186de21f..0000000000000000000000000000000000000000 --- a/tests/system-test/forcedrop +++ /dev/null @@ -1 +0,0 @@ -/home/ubuntu/Documents/github/cadem/emptydebug/forcedrop/ \ No newline at end of file