Merge pull request #20674 from taosdata/fix/TS-3010

fix: query schema is old issue

include/common/tdatablock.h

@@ -178,7 +178,7 @@ int32_t getJsonValueLen(const char* data);
 
 int32_t colDataSetVal(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
 int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, bool isNull);
-int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, uint32_t numOfRows);
+int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t rowIndex, const char* pData, uint32_t numOfRows, bool trimValue);
 int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, int32_t* capacity,
                         const SColumnInfoData* pSource, int32_t numOfRow2);
 int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows,

source/common/src/tdatablock.c

@@ -147,9 +147,17 @@ int32_t colDataReserve(SColumnInfoData* pColumnInfoData, size_t newSize) {
   return TSDB_CODE_SUCCESS;
 }
 
-static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t currentRow, const char* pData,
-                         int32_t itemLen, int32_t numOfRows) {
-  ASSERT(pColumnInfoData->info.bytes >= itemLen);
+static int32_t doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t currentRow, const char* pData,
+                         int32_t itemLen, int32_t numOfRows, bool trimValue) {
+  if (pColumnInfoData->info.bytes < itemLen) {
+    uWarn("column/tag actual data len %d is bigger than schema len %d, trim it:%d", itemLen, pColumnInfoData->info.bytes, trimValue);
+    if (trimValue) {
+      itemLen = pColumnInfoData->info.bytes;
+    } else {
+      return TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;
+    }
+  }
+  
   size_t start = 1;
 
   // the first item
@@ -178,10 +186,12 @@ static void doCopyNItems(struct SColumnInfoData* pColumnInfoData, int32_t curren
 
     pColumnInfoData->varmeta.length += numOfRows * itemLen;
   }
+
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData,
-                            uint32_t numOfRows) {
+                            uint32_t numOfRows, bool trimValue) {
   int32_t len = pColumnInfoData->info.bytes;
   if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
     len = varDataTLen(pData);
@@ -193,8 +203,7 @@ int32_t colDataSetNItems(SColumnInfoData* pColumnInfoData, uint32_t currentRow,
     }
   }
 
-  doCopyNItems(pColumnInfoData, currentRow, pData, len, numOfRows);
-  return TSDB_CODE_SUCCESS;
+  return doCopyNItems(pColumnInfoData, currentRow, pData, len, numOfRows, trimValue);
 }
 
 static void doBitmapMerge(SColumnInfoData* pColumnInfoData, int32_t numOfRow1, const SColumnInfoData* pSource,

source/dnode/vnode/inc/vnode.h

@@ -182,7 +182,7 @@ int32_t tsdbReaderOpen(SVnode *pVnode, SQueryTableDataCond *pCond, void *pTableL
 
 void         tsdbReaderSetId(STsdbReader* pReader, const char* idstr);
 void         tsdbReaderClose(STsdbReader *pReader);
-bool         tsdbNextDataBlock(STsdbReader *pReader);
+int32_t      tsdbNextDataBlock(STsdbReader *pReader, bool *hasNext);
 int32_t      tsdbRetrieveDatablockSMA(STsdbReader *pReader, SSDataBlock *pDataBlock, bool *allHave);
 void         tsdbReleaseDataBlock(STsdbReader *pReader);
 SSDataBlock *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList);

source/libs/executor/src/scanoperator.c

@@ -584,10 +584,16 @@ int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int
       if (isNullVal) {
         colDataSetNNULL(pColInfoData, 0, pBlock->info.rows);
       } else if (pColInfoData->info.type != TSDB_DATA_TYPE_JSON) {
-        colDataSetNItems(pColInfoData, 0, data, pBlock->info.rows);
+        code = colDataSetNItems(pColInfoData, 0, data, pBlock->info.rows, false);
         if (IS_VAR_DATA_TYPE(((const STagVal*)p)->type)) {
           taosMemoryFree(data);
         }
+        if (code) {
+          if (freeReader) {
+            metaReaderClear(&mr);
+          }
+          return code;
+        }
       } else {  // todo opt for json tag
         for (int32_t i = 0; i < pBlock->info.rows; ++i) {
           colDataSetVal(pColInfoData, i, data, false);
@@ -634,10 +640,22 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
   STableScanInfo* pTableScanInfo = pOperator->info;
   SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
   SSDataBlock*    pBlock = pTableScanInfo->pResBlock;
+  bool            hasNext = false;
+  int32_t         code = TSDB_CODE_SUCCESS;
 
   int64_t st = taosGetTimestampUs();
 
-  while (tsdbNextDataBlock(pTableScanInfo->base.dataReader)) {
+  while (true) {
+    code = tsdbNextDataBlock(pTableScanInfo->base.dataReader, &hasNext);
+    if (code) {
+      tsdbReleaseDataBlock(pTableScanInfo->base.dataReader);
+      T_LONG_JMP(pTaskInfo->env, code);
+    }
+
+    if (!hasNext) {
+      break;
+    }
+    
     if (isTaskKilled(pTaskInfo)) {
       tsdbReleaseDataBlock(pTableScanInfo->base.dataReader);
       T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
@@ -1015,7 +1033,15 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
     return NULL;
   }
 
-  if (tsdbNextDataBlock(pReader)) {
+  bool hasNext = false;
+  code = tsdbNextDataBlock(pReader, &hasNext);
+  if (code != TSDB_CODE_SUCCESS) {
+    terrno = code;
+    T_LONG_JMP(pTaskInfo->env, code);
+    return NULL;
+  }
+
+  if (hasNext) {
     /*SSDataBlock* p = */ tsdbRetrieveDataBlock(pReader, NULL);
     doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows);
     pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid);
@@ -2104,12 +2130,22 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
   // NOTE: this operator does never check if current status is done or not
   SExecTaskInfo*      pTaskInfo = pOperator->pTaskInfo;
   SStreamRawScanInfo* pInfo = pOperator->info;
+  int32_t             code = TSDB_CODE_SUCCESS;
   pTaskInfo->streamInfo.metaRsp.metaRspLen = 0;  // use metaRspLen !=0 to judge if data is meta
   pTaskInfo->streamInfo.metaRsp.metaRsp = NULL;
 
   qDebug("tmqsnap doRawScan called");
   if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
-    if (pInfo->dataReader && tsdbNextDataBlock(pInfo->dataReader)) {
+    bool hasNext = false;
+    if (pInfo->dataReader) {
+      code = tsdbNextDataBlock(pInfo->dataReader, &hasNext);
+      if (code) {
+        tsdbReleaseDataBlock(pInfo->dataReader);
+        longjmp(pTaskInfo->env, code);
+      }
+    }
+    
+    if (pInfo->dataReader && hasNext) {
       if (isTaskKilled(pTaskInfo)) {
         tsdbReleaseDataBlock(pInfo->dataReader);
         longjmp(pTaskInfo->env, pTaskInfo->code);
@@ -2610,8 +2646,21 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
   
   pInfo->base.dataReader = source->dataReader;
   STsdbReader* reader = pInfo->base.dataReader;
+  bool hasNext = false;
   qTrace("tsdb/read-table-data: %p, enter next reader", reader);
-  while (tsdbNextDataBlock(reader)) {
+
+  while (true) {
+    code = tsdbNextDataBlock(reader, &hasNext);
+    if (code != 0) {
+      tsdbReleaseDataBlock(reader);
+      pInfo->base.dataReader = NULL;
+      T_LONG_JMP(pTaskInfo->env, code);
+    }
+
+    if (!hasNext) {
+      break;
+    }
+  
     if (isTaskKilled(pTaskInfo)) {
       tsdbReleaseDataBlock(reader);
       pInfo->base.dataReader = NULL;

source/libs/executor/src/sysscanoperator.c

@@ -1627,7 +1627,7 @@ static void sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScan
     char             varTbName[TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE] = {0};
     STR_TO_VARSTR(varTbName, name);
 
-    colDataSetNItems(pColumnInfoData, 0, varTbName, pBlock->info.rows);
+    colDataSetNItems(pColumnInfoData, 0, varTbName, pBlock->info.rows, true);
   }
 
   doFilter(pBlock, pOperator->exprSupp.pFilterInfo, NULL);

source/libs/scalar/src/sclfunc.c

@@ -1755,7 +1755,11 @@ int32_t winEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
 int32_t qTbnameFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
   char* p = colDataGetVarData(pInput->columnData, 0);
 
-  colDataSetNItems(pOutput->columnData, pOutput->numOfRows, p, pInput->numOfRows);
+  int32_t code = colDataSetNItems(pOutput->columnData, pOutput->numOfRows, p, pInput->numOfRows, true);
+  if (code) {
+    return code;
+  }
+  
   pOutput->numOfRows += pInput->numOfRows;
   return TSDB_CODE_SUCCESS;
 }

tests/parallel_test/cases.task

@@ -14,6 +14,8 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_time.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_str.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py -Q 3
@@ -22,6 +24,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_str.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_time.py -Q 4
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqShow.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStb.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeStb0.py

tests/system-test/2-query/columnLenUpdated.py

+
+import taos
+import sys
+import time
+import socket
+import os
+import platform
+if platform.system().lower() == 'windows':
+    import wexpect as taosExpect
+else:
+    import pexpect as taosExpect
+
+from util.log import *
+from util.sql import *
+from util.cases import *
+from util.dnodes import *
+
+def taos_command (buildPath, key, value, expectString, sqlString=''):
+    if len(key) == 0:
+        tdLog.exit("taos test key is null!")
+
+    if platform.system().lower() == 'windows':
+        taosCmd = buildPath + '\\build\\bin\\taos.exe '
+        taosCmd = taosCmd.replace('\\','\\\\')
+    else:
+        taosCmd = buildPath + '/build/bin/taos '
+
+    cfgPath = buildPath + "/../sim/psim/cfg"
+    taosCmd = taosCmd + ' -c' + cfgPath + ' -' + key
+    if len(value) != 0:
+        taosCmd = taosCmd + ' ' + value
+
+    tdLog.info ("taos cmd: %s" % taosCmd)
+
+    child = taosExpect.spawn(taosCmd, timeout=20)
+    #output = child.readline()
+    #print (output.decode())
+    if len(expectString) != 0:
+        i = child.expect([expectString, taosExpect.TIMEOUT, taosExpect.EOF], timeout=20)
+    else:
+        i = child.expect([taosExpect.TIMEOUT, taosExpect.EOF], timeout=20)
+
+    if platform.system().lower() == 'windows':
+        retResult = child.before
+    else:
+        retResult = child.before.decode()
+    print(retResult)
+    #print(child.after.decode())
+    if i == 0:
+        print ('taos login success! Here can run sql, taos> ')
+        return  "TAOS_OK"
+    else:
+        return "TAOS_FAIL"
+
+class TDTestCase:
+    #updatecfgDict = {'clientCfg': {'serverPort': 7080, 'firstEp': 'trd02:7080', 'secondEp':'trd02:7080'},\
+    #                 'serverPort': 7080, 'firstEp': 'trd02:7080'}
+    hostname = socket.gethostname()
+    if (platform.system().lower() == 'windows' and not tdDnodes.dnodes[0].remoteIP == ""):
+        try:
+            config = eval(tdDnodes.dnodes[0].remoteIP)
+            hostname = config["host"]
+        except Exception:
+            hostname = tdDnodes.dnodes[0].remoteIP
+    serverPort = '7080'
+    rpcDebugFlagVal = '143'
+    clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
+    clientCfgDict["serverPort"]    = serverPort
+    clientCfgDict["firstEp"]       = hostname + ':' + serverPort
+    clientCfgDict["secondEp"]      = hostname + ':' + serverPort
+    clientCfgDict["rpcDebugFlag"]  = rpcDebugFlagVal
+    clientCfgDict["fqdn"] = hostname
+
+    updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
+    updatecfgDict["clientCfg"]  = clientCfgDict
+    updatecfgDict["serverPort"] = serverPort
+    updatecfgDict["firstEp"]    = hostname + ':' + serverPort
+    updatecfgDict["secondEp"]   = hostname + ':' + serverPort
+    updatecfgDict["fqdn"] = hostname
+
+    print ("===================: ", updatecfgDict)
+
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug(f"start to excute {__file__}")
+        tdSql.init(conn.cursor())
+
+    def getBuildPath(self):
+        selfPath = os.path.dirname(os.path.realpath(__file__))
+
+        if ("community" in selfPath):
+            projPath = selfPath[:selfPath.find("community")]
+        else:
+            projPath = selfPath[:selfPath.find("tests")]
+
+        for root, dirs, files in os.walk(projPath):
+            if ("taosd" in files or "taosd.exe" in files):
+                rootRealPath = os.path.dirname(os.path.realpath(root))
+                if ("packaging" not in rootRealPath):
+                    buildPath = root[:len(root) - len("/build/bin")]
+                    break
+        return buildPath
+
+    def run(self):  # sourcery skip: extract-duplicate-method, remove-redundant-fstring
+        tdSql.prepare()
+        # time.sleep(2)
+        tdSql.query("create user testpy pass 'testpy'")
+
+        buildPath = self.getBuildPath()
+        if (buildPath == ""):
+            tdLog.exit("taosd not found!")
+        else:
+            tdLog.info("taosd found in %s" % buildPath)
+        cfgPath = buildPath + "/../sim/psim/cfg"
+        tdLog.info("cfgPath: %s" % cfgPath)
+
+        checkNetworkStatus = ['0: unavailable', '1: network ok', '2: service ok', '3: service degraded', '4: exiting']
+        netrole            = ['client', 'server']
+
+        keyDict = {'h':'', 'P':'6030', 'p':'testpy', 'u':'testpy', 'a':'', 'A':'', 'c':'', 'C':'', 's':'', 'r':'', 'f':'', \
+                   'k':'', 't':'', 'n':'', 'l':'1024', 'N':'100', 'V':'', 'd':'db', 'w':'30', '-help':'', '-usage':'', '?':''}
+
+        keyDict['h'] = self.hostname
+        keyDict['c'] = cfgPath
+        keyDict['P'] = self.serverPort
+
+        tdSql.query("drop database if exists db1")
+        tdSql.query("create database if not exists db1 vgroups 1")
+        tdSql.query("use db1")
+        tdSql.query("create table tba (ts timestamp, f1 binary(2))")
+        tdSql.query("insert into tba values (now, '22')")
+        tdSql.query("select * from tba")
+        tdSql.checkData(0, 1, '22')
+
+        keyDict['s'] = "\"alter table db1.tba modify column f1 binary(5) \""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Query OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        keyDict['s'] = "\"insert into db1.tba values (now, '55555')\""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Insert OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        tdSql.query("select * from tba order by ts")
+        tdSql.checkData(0, 1, '22')
+        tdSql.checkData(1, 1, '55555')
+
+
+        tdSql.query("create table stb (ts timestamp, f1 int) tags (tg1 binary(2))")
+        tdSql.query("create table tb1 using stb tags('bb')")
+        tdSql.query("insert into tb1 values (now, 2)")
+        tdSql.query("select count(*) from stb group by tg1")
+        tdSql.checkData(0, 0, 1)
+
+        keyDict['s'] = "\"alter table db1.stb modify tag tg1 binary(5) \""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Query OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        keyDict['s'] = "\"create table db1.tb2 using db1.stb tags('bbbbb')\""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Create OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        keyDict['s'] = "\"insert into db1.tb2 values (now, 2)\""
+        retCode = taos_command(buildPath, "s", keyDict['s'], "Insert OK", '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
+
+        tdSql.query("select count(*) from stb group by tg1")
+        tdSql.checkData(0, 0, 1)
+        tdSql.checkData(1, 0, 1)
+
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success(f"{__file__} successfully executed")
+
+tdCases.addLinux(__file__, TDTestCase())
+tdCases.addWindows(__file__, TDTestCase())