Merge branch '3.0' into feature/TD-13066-3.0

include/util/taoserror.h

@@ -419,6 +419,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TQ_META_KEY_NOT_IN_TXN        TAOS_DEF_ERROR_CODE(0, 0x0A09)
 #define TSDB_CODE_TQ_META_KEY_DUP_IN_TXN        TAOS_DEF_ERROR_CODE(0, 0x0A0A)
 #define TSDB_CODE_TQ_GROUP_NOT_SET              TAOS_DEF_ERROR_CODE(0, 0x0A0B)
+#define TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND     TAOS_DEF_ERROR_CODE(0, 0x0A0B)
 
 // wal
 #define TSDB_CODE_WAL_APP_ERROR                 TAOS_DEF_ERROR_CODE(0, 0x1000)

source/dnode/vnode/src/tq/tq.c

@@ -600,6 +600,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
           SSDataBlock block = {0};
           if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.uid, &block.info.rows,
                                   &block.info.numOfCols) < 0) {
+            if (terrno == TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND) continue;
             ASSERT(0);
           }
           int32_t            dataStrLen = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(&block);

source/dnode/vnode/src/tq/tqRead.c

@@ -91,12 +91,22 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
   if (pHandle->sver != sversion || pHandle->cachedSchemaUid != pHandle->msgIter.suid) {
     pHandle->pSchema = metaGetTbTSchema(pHandle->pVnodeMeta, pHandle->msgIter.uid, sversion);
     if (pHandle->pSchema == NULL) {
-      tqError("cannot found schema for table: %ld, version %d", pHandle->msgIter.suid, pHandle->sver);
+      tqWarn("cannot found tsschema for table: uid: %ld (suid: %ld), version %d, possibly dropped table",
+             pHandle->msgIter.uid, pHandle->msgIter.suid, pHandle->sver);
+      /*ASSERT(0);*/
+      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
       return -1;
     }
 
     // this interface use suid instead of uid
     pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, pHandle->msgIter.suid, sversion, true);
+    if (pHandle->pSchemaWrapper == NULL) {
+      tqWarn("cannot found schema wrapper for table: suid: %ld, version %d, possibly dropped table",
+             pHandle->msgIter.suid, pHandle->sver);
+      /*ASSERT(0);*/
+      terrno = TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND;
+      return -1;
+    }
     pHandle->sver = sversion;
     pHandle->cachedSchemaUid = pHandle->msgIter.suid;
   }