xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/3.0' into feature/sync2-merge

This commit is contained in:
Shengliang Guan 2022-10-23 14:40:27 +08:00
parent 1382242e17 e86cdc8cde
commit fbf354a78c
18 changed files with 1304 additions and 279 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
source/client/src/clientTmqConnector.c
View File

@ -73,8 +73,8 @@ JNIEXPORT void JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqConfDestroyImp
if (conf == NULL) { if (conf == NULL) {
jniDebug("jobj:%p, tmq config is already destroyed", jobj); jniDebug("jobj:%p, tmq config is already destroyed", jobj);
} else { } else {
tmq_conf_destroy(conf);
jniDebug("jobj:%p, config:%p, tmq successfully destroy config", jobj, conf); jniDebug("jobj:%p, config:%p, tmq successfully destroy config", jobj, conf);
tmq_conf_destroy(conf);
} }
} }
@ -206,6 +206,7 @@ JNIEXPORT void JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqCommitAsync(JN
tmq_t *tmq = (tmq_t *)jtmq; tmq_t *tmq = (tmq_t *)jtmq;
if (tmq == NULL) { if (tmq == NULL) {
jniError("jobj:%p, tmq is closed", jobj); jniError("jobj:%p, tmq is closed", jobj);
return;
} }
TAOS_RES *res = (TAOS_RES *)jres; TAOS_RES *res = (TAOS_RES *)jres;
consumer = (*env)->NewGlobalRef(env, consumer); consumer = (*env)->NewGlobalRef(env, consumer);
@ -252,6 +253,7 @@ JNIEXPORT jstring JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqGetTopicNam
TAOS_RES *res = (TAOS_RES *)jres; TAOS_RES *res = (TAOS_RES *)jres;
if (res == NULL) { if (res == NULL) {
jniDebug("jobj:%p, invalid res handle", jobj); jniDebug("jobj:%p, invalid res handle", jobj);
return NULL;
} }
return (*env)->NewStringUTF(env, tmq_get_topic_name(res)); return (*env)->NewStringUTF(env, tmq_get_topic_name(res));
} }
@ -259,6 +261,7 @@ JNIEXPORT jstring JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqGetDbName(J
TAOS_RES *res = (TAOS_RES *)jres; TAOS_RES *res = (TAOS_RES *)jres;
if (res == NULL) { if (res == NULL) {
jniDebug("jobj:%p, invalid res handle", jobj); jniDebug("jobj:%p, invalid res handle", jobj);
return NULL;
} }
return (*env)->NewStringUTF(env, tmq_get_db_name(res)); return (*env)->NewStringUTF(env, tmq_get_db_name(res));
} }
@ -266,6 +269,7 @@ JNIEXPORT jint JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqGetVgroupId(JN
TAOS_RES *res = (TAOS_RES *)jres; TAOS_RES *res = (TAOS_RES *)jres;
if (res == NULL) { if (res == NULL) {
jniDebug("jobj:%p, invalid res handle", jobj); jniDebug("jobj:%p, invalid res handle", jobj);
return -1;
} }
return tmq_get_vgroup_id(res); return tmq_get_vgroup_id(res);
} }
@ -275,6 +279,7 @@ JNIEXPORT jstring JNICALL Java_com_taosdata_jdbc_tmq_TMQConnector_tmqGetTableNam
TAOS_RES *res = (TAOS_RES *)jres; TAOS_RES *res = (TAOS_RES *)jres;
if (res == NULL) { if (res == NULL) {
jniDebug("jobj:%p, invalid res handle", jobj); jniDebug("jobj:%p, invalid res handle", jobj);
return NULL;
} }
return (*env)->NewStringUTF(env, tmq_get_table_name(res)); return (*env)->NewStringUTF(env, tmq_get_table_name(res));
} }

4
source/dnode/vnode/inc/vnode.h
View File

@ -120,7 +120,11 @@ typedef struct SMetaFltParam {
} SMetaFltParam; } SMetaFltParam;
// TODO, refactor later
int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *results); int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *results);
int32_t metaFilterCreateTime(SMeta *pMeta, SMetaFltParam *parm, SArray *pUids);
int32_t metaFilterTableName(SMeta *pMeta, SMetaFltParam *param, SArray *pUids);
int32_t metaFilterTtl(SMeta *pMeta, SMetaFltParam *param, SArray *pUids);
#if 1 // refact APIs below (TODO) #if 1 // refact APIs below (TODO)
typedef SVCreateTbReq STbCfg; typedef SVCreateTbReq STbCfg;

14
source/dnode/vnode/src/inc/meta.h
View File

@ -86,9 +86,13 @@ struct SMeta {
TTB* pSuidIdx; TTB* pSuidIdx;
// ivt idx and idx // ivt idx and idx
void* pTagIvtIdx; void* pTagIvtIdx;
TTB* pTagIdx; TTB* pTagIdx;
TTB* pTtlIdx; TTB* pTtlIdx;
TTB* pCtimeIdx; // table created time idx
TTB* pNcolIdx; // ncol of table idx, normal table only
TTB* pSmaIdx; TTB* pSmaIdx;
// stream // stream
@ -142,6 +146,16 @@ typedef struct {
int64_t smaUid; int64_t smaUid;
} SSmaIdxKey; } SSmaIdxKey;
typedef struct {
int64_t ctime;
tb_uid_t uid;
} SCtimeIdxKey;
typedef struct {
int64_t ncol;
tb_uid_t uid;
} SNcolIdxKey;
// metaTable ================== // metaTable ==================
int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void* pTagData, int32_t nTagData, int8_t type, tb_uid_t uid, int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void* pTagData, int32_t nTagData, int8_t type, tb_uid_t uid,
STagIdxKey** ppTagIdxKey, int32_t* nTagIdxKey); STagIdxKey** ppTagIdxKey, int32_t* nTagIdxKey);

58
source/dnode/vnode/src/meta/metaOpen.c
View File

@ -24,6 +24,9 @@ static int uidIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kL
static int smaIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); static int smaIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
static int taskIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); static int taskIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
static int ctimeIdxCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
static int ncolIdxCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
static int32_t metaInitLock(SMeta *pMeta) { return taosThreadRwlockInit(&pMeta->lock, NULL); } static int32_t metaInitLock(SMeta *pMeta) { return taosThreadRwlockInit(&pMeta->lock, NULL); }
static int32_t metaDestroyLock(SMeta *pMeta) { return taosThreadRwlockDestroy(&pMeta->lock); } static int32_t metaDestroyLock(SMeta *pMeta) { return taosThreadRwlockDestroy(&pMeta->lock); }
@ -139,6 +142,20 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
goto _err; goto _err;
} }
// idx table create time
ret = tdbTbOpen("ctime.idx", sizeof(SCtimeIdxKey), 0, ctimeIdxCmpr, pMeta->pEnv, &pMeta->pCtimeIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta ctime index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
// idx num of col, normal table only
ret = tdbTbOpen("ncol.idx", sizeof(SNcolIdxKey), 0, ncolIdxCmpr, pMeta->pEnv, &pMeta->pNcolIdx, 0);
if (ret < 0) {
metaError("vgId:%d, failed to open meta ncol index since %s", TD_VID(pVnode), tstrerror(terrno));
goto _err;
}
ret = tdbTbOpen("stream.task.db", sizeof(int64_t), -1, taskIdxKeyCmpr, pMeta->pEnv, &pMeta->pStreamDb, 0); ret = tdbTbOpen("stream.task.db", sizeof(int64_t), -1, taskIdxKeyCmpr, pMeta->pEnv, &pMeta->pStreamDb, 0);
if (ret < 0) { if (ret < 0) {
metaError("vgId:%d, failed to open meta stream task index since %s", TD_VID(pVnode), tstrerror(terrno)); metaError("vgId:%d, failed to open meta stream task index since %s", TD_VID(pVnode), tstrerror(terrno));
@ -166,6 +183,8 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) {
_err: _err:
if (pMeta->pIdx) metaCloseIdx(pMeta); if (pMeta->pIdx) metaCloseIdx(pMeta);
if (pMeta->pStreamDb) tdbTbClose(pMeta->pStreamDb); if (pMeta->pStreamDb) tdbTbClose(pMeta->pStreamDb);
if (pMeta->pNcolIdx) tdbTbClose(pMeta->pNcolIdx);
if (pMeta->pCtimeIdx) tdbTbClose(pMeta->pCtimeIdx);
if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx); if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx);
if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx); if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx);
if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx); if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx);
@ -187,6 +206,8 @@ int metaClose(SMeta *pMeta) {
if (pMeta->pCache) metaCacheClose(pMeta); if (pMeta->pCache) metaCacheClose(pMeta);
if (pMeta->pIdx) metaCloseIdx(pMeta); if (pMeta->pIdx) metaCloseIdx(pMeta);
if (pMeta->pStreamDb) tdbTbClose(pMeta->pStreamDb); if (pMeta->pStreamDb) tdbTbClose(pMeta->pStreamDb);
if (pMeta->pNcolIdx) tdbTbClose(pMeta->pNcolIdx);
if (pMeta->pCtimeIdx) tdbTbClose(pMeta->pCtimeIdx);
if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx); if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx);
if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx); if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx);
if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx); if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx);
@ -391,6 +412,43 @@ static int ttlIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kL
return 0; return 0;
} }
static int ctimeIdxCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) {
SCtimeIdxKey *pCtimeIdxKey1 = (SCtimeIdxKey *)pKey1;
SCtimeIdxKey *pCtimeIdxKey2 = (SCtimeIdxKey *)pKey2;
if (pCtimeIdxKey1->ctime > pCtimeIdxKey2->ctime) {
return 1;
} else if (pCtimeIdxKey1->ctime < pCtimeIdxKey2->ctime) {
return -1;
}
if (pCtimeIdxKey1->uid > pCtimeIdxKey2->uid) {
return 1;
} else if (pCtimeIdxKey1->uid < pCtimeIdxKey2->uid) {
return -1;
}
return 0;
}
static int ncolIdxCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) {
SNcolIdxKey *pNcolIdxKey1 = (SNcolIdxKey *)pKey1;
SNcolIdxKey *pNcolIdxKey2 = (SNcolIdxKey *)pKey2;
if (pNcolIdxKey1->ncol > pNcolIdxKey2->ncol) {
return 1;
} else if (pNcolIdxKey1->ncol < pNcolIdxKey2->ncol) {
return -1;
}
if (pNcolIdxKey1->uid > pNcolIdxKey2->uid) {
return 1;
} else if (pNcolIdxKey1->uid < pNcolIdxKey2->uid) {
return -1;
}
return 0;
}
static int smaIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) { static int smaIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) {
SSmaIdxKey *pSmaIdxKey1 = (SSmaIdxKey *)pKey1; SSmaIdxKey *pSmaIdxKey1 = (SSmaIdxKey *)pKey1;
SSmaIdxKey *pSmaIdxKey2 = (SSmaIdxKey *)pKey2; SSmaIdxKey *pSmaIdxKey2 = (SSmaIdxKey *)pKey2;

144
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -1038,6 +1038,143 @@ typedef struct {
int32_t vLen; int32_t vLen;
} SIdxCursor; } SIdxCursor;
int32_t metaFilterCreateTime(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t ret = 0;
SIdxCursor *pCursor = NULL;
pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));
pCursor->pMeta = pMeta;
pCursor->suid = param->suid;
pCursor->cid = param->cid;
pCursor->type = param->type;
metaRLock(pMeta);
ret = tdbTbcOpen(pMeta->pCtimeIdx, &pCursor->pCur, NULL);
if (ret != 0) {
goto END;
}
int64_t uidLimit = param->reverse ? INT64_MAX : 0;
SCtimeIdxKey ctimeKey = {.ctime = *(int64_t *)(param->val), .uid = uidLimit};
SCtimeIdxKey *pCtimeKey = &ctimeKey;
int cmp = 0;
if (tdbTbcMoveTo(pCursor->pCur, &ctimeKey, sizeof(ctimeKey), &cmp) < 0) {
goto END;
}
bool first = true;
int32_t valid = 0;
while (1) {
void *entryKey = NULL;
int32_t nEntryKey = -1;
valid = tdbTbcGet(pCursor->pCur, (const void **)&entryKey, &nEntryKey, NULL, NULL);
if (valid < 0) break;
SCtimeIdxKey *p = entryKey;
int32_t cmp = (*param->filterFunc)((void *)&p->ctime, (void *)&pCtimeKey->ctime, param->type);
if (cmp == 0) taosArrayPush(pUids, &p->uid);
if (cmp == -1) break;
valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur);
if (valid < 0) break;
}
END:
if (pCursor->pMeta) metaULock(pCursor->pMeta);
if (pCursor->pCur) tdbTbcClose(pCursor->pCur);
taosMemoryFree(pCursor);
return ret;
}
int32_t metaFilterTableName(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t ret = 0;
char *buf = NULL;
STagIdxKey *pKey = NULL;
int32_t nKey = 0;
SIdxCursor *pCursor = NULL;
pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));
pCursor->pMeta = pMeta;
pCursor->suid = param->suid;
pCursor->cid = param->cid;
pCursor->type = param->type;
char *pName = param->val;
metaRLock(pMeta);
ret = tdbTbcOpen(pMeta->pNameIdx, &pCursor->pCur, NULL);
if (ret != 0) {
goto END;
}
int cmp = 0;
if (tdbTbcMoveTo(pCursor->pCur, pName, strlen(pName) + 1, &cmp) < 0) {
goto END;
}
bool first = true;
int32_t valid = 0;
while (1) {
void *pEntryKey = NULL, *pEntryVal = NULL;
int32_t nEntryKey = -1, nEntryVal = 0;
valid = tdbTbcGet(pCursor->pCur, (const void **)pEntryKey, &nEntryKey, (const void **)&pEntryVal, &nEntryVal);
if (valid < 0) break;
char *pTableKey = (char *)pEntryKey;
int32_t cmp = (*param->filterFunc)(pTableKey, pName, pCursor->type);
if (cmp == 0) {
tb_uid_t tuid = *(tb_uid_t *)pEntryVal;
taosArrayPush(pUids, &tuid);
} else if (cmp == 1) {
// next
} else {
break;
}
valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur);
if (valid < 0) {
break;
}
}
END:
if (pCursor->pMeta) metaULock(pCursor->pMeta);
if (pCursor->pCur) tdbTbcClose(pCursor->pCur);
taosMemoryFree(buf);
taosMemoryFree(pKey);
taosMemoryFree(pCursor);
return ret;
}
int32_t metaFilterTtl(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t ret = 0;
char *buf = NULL;
STtlIdxKey *pKey = NULL;
int32_t nKey = 0;
SIdxCursor *pCursor = NULL;
pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));
pCursor->pMeta = pMeta;
pCursor->suid = param->suid;
pCursor->cid = param->cid;
pCursor->type = param->type;
metaRLock(pMeta);
ret = tdbTbcOpen(pMeta->pTtlIdx, &pCursor->pCur, NULL);
END:
if (pCursor->pMeta) metaULock(pCursor->pMeta);
if (pCursor->pCur) tdbTbcClose(pCursor->pCur);
taosMemoryFree(buf);
taosMemoryFree(pKey);
taosMemoryFree(pCursor);
return ret;
// impl later
return 0;
}
int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) { int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
int32_t ret = 0; int32_t ret = 0;
char *buf = NULL; char *buf = NULL;
@ -1053,7 +1190,7 @@ int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
pCursor->type = param->type; pCursor->type = param->type;
metaRLock(pMeta); metaRLock(pMeta);
ret = tdbTbcOpen(pMeta->pTagIdx, &pCursor->pCur, NULL); ret = tdbTbcOpen(pMeta->pCtimeIdx, &pCursor->pCur, NULL);
if (ret < 0) { if (ret < 0) {
goto END; goto END;
} }
@ -1064,6 +1201,7 @@ int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
if (param->val == NULL) { if (param->val == NULL) {
metaError("vgId:%d, failed to filter NULL data", TD_VID(pMeta->pVnode)); metaError("vgId:%d, failed to filter NULL data", TD_VID(pMeta->pVnode));
ret = -1;
goto END; goto END;
} else { } else {
if (IS_VAR_DATA_TYPE(param->type)) { if (IS_VAR_DATA_TYPE(param->type)) {
@ -1104,9 +1242,11 @@ int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
valid = tdbTbcGet(pCursor->pCur, (const void **)&entryKey, &nEntryKey, (const void **)&entryVal, &nEntryVal); valid = tdbTbcGet(pCursor->pCur, (const void **)&entryKey, &nEntryKey, (const void **)&entryVal, &nEntryVal);
if (valid < 0) { if (valid < 0) {
tdbFree(entryVal);
break; break;
} }
STagIdxKey *p = entryKey; STagIdxKey *p = entryKey;
if (p == NULL) break;
if (p->type != pCursor->type) { if (p->type != pCursor->type) {
if (first) { if (first) {
valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur); valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur);
@ -1116,7 +1256,7 @@ int32_t metaFilterTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
break; break;
} }
} }
if (p == NULL || p->suid != pKey->suid) { if (p->suid != pKey->suid) {
break; break;
} }
first = false; first = false;

67
source/dnode/vnode/src/meta/metaTable.c
View File

@ -27,6 +27,11 @@ static int metaUpdateSuidIdx(SMeta *pMeta, const SMetaEntry *pME);
static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry); static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry);
static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type); static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type);
static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey); static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey);
// opt ins_tables query
static int metaUpdateCtimeIdx(SMeta *pMeta, const SMetaEntry *pME);
static int metaDeleteCtimeIdx(SMeta *pMeta, const SMetaEntry *pME);
static int metaUpdateNcolIdx(SMeta *pMeta, const SMetaEntry *pME);
static int metaDeleteNcolIdx(SMeta *pMeta, const SMetaEntry *pME);
static void metaGetEntryInfo(const SMetaEntry *pEntry, SMetaInfo *pInfo) { static void metaGetEntryInfo(const SMetaEntry *pEntry, SMetaInfo *pInfo) {
pInfo->uid = pEntry->uid; pInfo->uid = pEntry->uid;
@ -551,6 +556,26 @@ static void metaBuildTtlIdxKey(STtlIdxKey *ttlKey, const SMetaEntry *pME) {
ttlKey->dtime = ctime / 1000 + ttlDays * tsTtlUnit; ttlKey->dtime = ctime / 1000 + ttlDays * tsTtlUnit;
ttlKey->uid = pME->uid; ttlKey->uid = pME->uid;
} }
static int metaBuildCtimeIdxKey(SCtimeIdxKey *ctimeKey, const SMetaEntry *pME) {
int64_t ctime;
if (pME->type == TSDB_CHILD_TABLE) {
ctime = pME->ctbEntry.ctime;
} else if (pME->type == TSDB_NORMAL_TABLE) {
ctime = pME->ntbEntry.ctime;
} else {
return -1;
}
ctimeKey->ctime = ctime;
ctimeKey->uid = pME->uid;
return 0;
}
static int metaBuildNColIdxKey(SNcolIdxKey *ncolKey, const SMetaEntry *pME) {
ncolKey->ncol = pME->ntbEntry.schemaRow.nCols;
ncolKey->uid = pME->uid;
return 0;
}
static int metaDeleteTtlIdx(SMeta *pMeta, const SMetaEntry *pME) { static int metaDeleteTtlIdx(SMeta *pMeta, const SMetaEntry *pME) {
STtlIdxKey ttlKey = {0}; STtlIdxKey ttlKey = {0};
@ -632,6 +657,9 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
tdbTbDelete(pMeta->pNameIdx, e.name, strlen(e.name) + 1, &pMeta->txn); tdbTbDelete(pMeta->pNameIdx, e.name, strlen(e.name) + 1, &pMeta->txn);
tdbTbDelete(pMeta->pUidIdx, &uid, sizeof(uid), &pMeta->txn); tdbTbDelete(pMeta->pUidIdx, &uid, sizeof(uid), &pMeta->txn);
if (e.type == TSDB_CHILD_TABLE || e.type == TSDB_NORMAL_TABLE) metaDeleteCtimeIdx(pMeta, &e);
if (e.type == TSDB_NORMAL_TABLE) metaDeleteNcolIdx(pMeta, &e);
if (e.type != TSDB_SUPER_TABLE) metaDeleteTtlIdx(pMeta, &e); if (e.type != TSDB_SUPER_TABLE) metaDeleteTtlIdx(pMeta, &e);
if (e.type == TSDB_CHILD_TABLE) { if (e.type == TSDB_CHILD_TABLE) {
@ -658,6 +686,37 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) {
return 0; return 0;
} }
// opt ins_tables
int metaUpdateCtimeIdx(SMeta *pMeta, const SMetaEntry *pME) {
SCtimeIdxKey ctimeKey = {0};
if (metaBuildCtimeIdxKey(&ctimeKey, pME) < 0) {
return 0;
}
return tdbTbInsert(pMeta->pCtimeIdx, &ctimeKey, sizeof(ctimeKey), NULL, 0, &pMeta->txn);
}
int metaDeleteCtimeIdx(SMeta *pMeta, const SMetaEntry *pME) {
SCtimeIdxKey ctimeKey = {0};
if (metaBuildCtimeIdxKey(&ctimeKey, pME) < 0) {
return 0;
}
return tdbTbDelete(pMeta->pCtimeIdx, &ctimeKey, sizeof(ctimeKey), &pMeta->txn);
}
int metaUpdateNcolIdx(SMeta *pMeta, const SMetaEntry *pME) {
SNcolIdxKey ncolKey = {0};
if (metaBuildNColIdxKey(&ncolKey, pME) < 0) {
return 0;
}
return tdbTbInsert(pMeta->pNcolIdx, &ncolKey, sizeof(ncolKey), NULL, 0, &pMeta->txn);
}
int metaDeleteNcolIdx(SMeta *pMeta, const SMetaEntry *pME) {
SNcolIdxKey ncolKey = {0};
if (metaBuildNColIdxKey(&ncolKey, pME) < 0) {
return 0;
}
return tdbTbDelete(pMeta->pNcolIdx, &ncolKey, sizeof(ncolKey), &pMeta->txn);
}
static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq, STableMetaRsp *pMetaRsp) { static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq, STableMetaRsp *pMetaRsp) {
void *pVal = NULL; void *pVal = NULL;
@ -1278,12 +1337,14 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
ret = metaSaveJsonVarToIdx(pMeta, pCtbEntry, pTagColumn); ret = metaSaveJsonVarToIdx(pMeta, pCtbEntry, pTagColumn);
goto end; goto end;
} }
if (pTagData != NULL) {
if (metaCreateTagIdxKey(pCtbEntry->ctbEntry.suid, pTagColumn->colId, pTagData, nTagData, pTagColumn->type, if (metaCreateTagIdxKey(pCtbEntry->ctbEntry.suid, pTagColumn->colId, pTagData, nTagData, pTagColumn->type,
pCtbEntry->uid, &pTagIdxKey, &nTagIdxKey) < 0) { pCtbEntry->uid, &pTagIdxKey, &nTagIdxKey) < 0) {
ret = -1; ret = -1;
goto end; goto end;
} }
tdbTbUpsert(pMeta->pTagIdx, pTagIdxKey, nTagIdxKey, NULL, 0, &pMeta->txn); tdbTbUpsert(pMeta->pTagIdx, pTagIdxKey, nTagIdxKey, NULL, 0, &pMeta->txn);
}
end: end:
metaDestroyTagIdxKey(pTagIdxKey); metaDestroyTagIdxKey(pTagIdxKey);
tDecoderClear(&dc); tDecoderClear(&dc);
@ -1370,6 +1431,12 @@ int metaHandleEntry(SMeta *pMeta, const SMetaEntry *pME) {
} }
} }
if (metaUpdateCtimeIdx(pMeta, pME) < 0) goto _err;
if (pME->type == TSDB_NORMAL_TABLE) {
if (metaUpdateNcolIdx(pMeta, pME) < 0) goto _err;
}
if (pME->type != TSDB_SUPER_TABLE) { if (pME->type != TSDB_SUPER_TABLE) {
if (metaUpdateTtlIdx(pMeta, pME) < 0) goto _err; if (metaUpdateTtlIdx(pMeta, pME) < 0) goto _err;
} }

9
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -1492,4 +1492,11 @@ void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity) {
size_t tsdbCacheGetCapacity(SVnode *pVnode) { return taosLRUCacheGetCapacity(pVnode->pTsdb->lruCache); } size_t tsdbCacheGetCapacity(SVnode *pVnode) { return taosLRUCacheGetCapacity(pVnode->pTsdb->lruCache); }
size_t tsdbCacheGetUsage(SVnode *pVnode) { return taosLRUCacheGetUsage(pVnode->pTsdb->lruCache); } size_t tsdbCacheGetUsage(SVnode *pVnode) {
size_t usage = 0;
if (pVnode->pTsdb != NULL) {
usage = taosLRUCacheGetUsage(pVnode->pTsdb->lruCache);
}
return usage;
}

7
source/libs/executor/inc/executorimpl.h
View File

@ -541,6 +541,12 @@ typedef struct {
SSnapContext* sContext; SSnapContext* sContext;
} SStreamRawScanInfo; } SStreamRawScanInfo;
typedef struct SSysTableIndex {
int8_t init;
SArray *uids;
int32_t lastIdx;
} SSysTableIndex;
typedef struct SSysTableScanInfo { typedef struct SSysTableScanInfo {
SRetrieveMetaTableRsp* pRsp; SRetrieveMetaTableRsp* pRsp;
SRetrieveTableReq req; SRetrieveTableReq req;
@ -553,6 +559,7 @@ typedef struct SSysTableScanInfo {
bool showRewrite; bool showRewrite;
SNode* pCondition; // db_name filter condition, to discard data that are not in current database SNode* pCondition; // db_name filter condition, to discard data that are not in current database
SMTbCursor* pCur; // cursor for iterate the local table meta store. SMTbCursor* pCur; // cursor for iterate the local table meta store.
SSysTableIndex* pIdx; // idx for local table meta
SArray* scanCols; // SArray<int16_t> scan column id list SArray* scanCols; // SArray<int16_t> scan column id list
SName name; SName name;
SSDataBlock* pRes; SSDataBlock* pRes;

15
source/libs/executor/src/executil.c
View File

@ -422,16 +422,6 @@ static SColumnInfoData* getColInfoResult(void* metaHandle, int64_t suid, SArray*
goto end; goto end;
} }
} }
/*else {
code = metaGetTableTagsByUids(metaHandle, suid, uidList, tags);
if (code != 0) {
terrno = code;
qError("failed to get table from meta idx, reason: %s, suid:%" PRId64, tstrerror(code), suid);
goto end;
} else {
qInfo("succ to get table from meta idx, suid:%" PRId64, suid);
}
}*/
int32_t rows = taosArrayGetSize(uidList); int32_t rows = taosArrayGetSize(uidList);
if (rows == 0) { if (rows == 0) {
@ -1216,7 +1206,6 @@ void createExprFromOneNode(SExprInfo* pExp, SNode* pNode, int16_t slotId) {
if (!pFuncNode->pParameterList && (memcmp(pExprNode->_function.functionName, name, len) == 0) && if (!pFuncNode->pParameterList && (memcmp(pExprNode->_function.functionName, name, len) == 0) &&
pExprNode->_function.functionName[len] == 0) { pExprNode->_function.functionName[len] == 0) {
pFuncNode->pParameterList = nodesMakeList(); pFuncNode->pParameterList = nodesMakeList();
ASSERT(LIST_LENGTH(pFuncNode->pParameterList) == 0); ASSERT(LIST_LENGTH(pFuncNode->pParameterList) == 0);
SValueNode* res = (SValueNode*)nodesMakeNode(QUERY_NODE_VALUE); SValueNode* res = (SValueNode*)nodesMakeNode(QUERY_NODE_VALUE);
@ -1266,8 +1255,8 @@ void createExprFromOneNode(SExprInfo* pExp, SNode* pNode, int16_t slotId) {
pExp->base.numOfParams = 1; pExp->base.numOfParams = 1;
SDataType* pType = &pCaseNode->node.resType; SDataType* pType = &pCaseNode->node.resType;
pExp->base.resSchema = createResSchema(pType->type, pType->bytes, slotId, pType->scale, pExp->base.resSchema =
pType->precision, pCaseNode->node.aliasName); createResSchema(pType->type, pType->bytes, slotId, pType->scale, pType->precision, pCaseNode->node.aliasName);
pExp->pExpr->_optrRoot.pRootNode = pNode; pExp->pExpr->_optrRoot.pRootNode = pNode;
} else { } else {
ASSERT(0); ASSERT(0);

735
source/libs/executor/src/scanoperator.c
View File

@ -39,6 +39,64 @@ static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capac
static int32_t buildDbTableInfoBlock(bool sysInfo, const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, static int32_t buildDbTableInfoBlock(bool sysInfo, const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta,
size_t size, const char* dbName); size_t size, const char* dbName);
static char* SYSTABLE_IDX_COLUMN[] = {"table_name", "db_name", "create_time", "columns",
"ttl", "stable_name", "vgroup_id', 'uid", "type"};
static char* SYSTABLE_IDX_EXCEPT[] = {"db_name", "vgroup_id"};
typedef int32_t (*__sys_filte)(void* pMeta, SNode* cond, SArray* result);
typedef int32_t (*__sys_check)(SNode* cond);
typedef struct {
const char* name;
__sys_check chkFunc;
__sys_filte fltFunc;
} SSTabFltFuncDef;
typedef struct {
void* pMeta;
void* pVnode;
} SSTabFltArg;
static int32_t sysChkFilter__Comm(SNode* pNode);
static int32_t sysChkFilter__DBName(SNode* pNode);
static int32_t sysChkFilter__VgroupId(SNode* pNode);
static int32_t sysChkFilter__TableName(SNode* pNode);
static int32_t sysChkFilter__CreateTime(SNode* pNode);
static int32_t sysChkFilter__Ncolumn(SNode* pNode);
static int32_t sysChkFilter__Ttl(SNode* pNode);
static int32_t sysChkFilter__STableName(SNode* pNode);
static int32_t sysChkFilter__Uid(SNode* pNode);
static int32_t sysChkFilter__Type(SNode* pNode);
static int32_t sysFilte__DbName(void* pMeta, SNode* pNode, SArray* result);
static int32_t sysFilte__VgroupId(void* pMeta, SNode* pNode, SArray* result);
static int32_t sysFilte__TableName(void* pMeta, SNode* pNode, SArray* result);
static int32_t sysFilte__CreateTime(void* pMeta, SNode* pNode, SArray* result);
static int32_t sysFilte__Ncolumn(void* pMeta, SNode* pNode, SArray* result);
static int32_t sysFilte__Ttl(void* pMeta, SNode* pNode, SArray* result);
static int32_t sysFilte__STableName(void* pMeta, SNode* pNode, SArray* result);
static int32_t sysFilte__Uid(void* pMeta, SNode* pNode, SArray* result);
static int32_t sysFilte__Type(void* pMeta, SNode* pNode, SArray* result);
const SSTabFltFuncDef filterDict[] = {
{.name = "table_name", .chkFunc = sysChkFilter__TableName, .fltFunc = sysFilte__TableName},
{.name = "db_name", .chkFunc = sysChkFilter__DBName, .fltFunc = sysFilte__DbName},
{.name = "create_time", .chkFunc = sysChkFilter__CreateTime, .fltFunc = sysFilte__CreateTime},
{.name = "columns", .chkFunc = sysChkFilter__Ncolumn, .fltFunc = sysFilte__Ncolumn},
{.name = "ttl", .chkFunc = sysChkFilter__Ttl, .fltFunc = sysFilte__Ttl},
{.name = "stable_name", .chkFunc = sysChkFilter__STableName, .fltFunc = sysFilte__STableName},
{.name = "vgroup_id", .chkFunc = sysChkFilter__VgroupId, .fltFunc = sysFilte__VgroupId},
{.name = "uid", .chkFunc = sysChkFilter__Uid, .fltFunc = sysFilte__Uid},
{.name = "type", .chkFunc = sysChkFilter__Type, .fltFunc = sysFilte__Type}};
#define SYSTAB_FILTER_DICT_SIZE (sizeof(filterDict) / sizeof(filterDict[0]))
static int32_t optSysTabFilte(void* arg, SNode* cond, SArray* result);
static int32_t optSysTabFilteImpl(void* arg, SNode* cond, SArray* result);
static int32_t optSysCheckOper(SNode* pOpear);
static int32_t optSysMergeRslt(SArray* multiRslt, SArray* reslt);
static bool processBlockWithProbability(const SSampleExecInfo* pInfo); static bool processBlockWithProbability(const SSampleExecInfo* pInfo);
static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo, SMetaReader* smrSuperTable, static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo, SMetaReader* smrSuperTable,
@ -2323,6 +2381,11 @@ static void destroySysScanOperator(void* param) {
metaCloseTbCursor(pInfo->pCur); metaCloseTbCursor(pInfo->pCur);
pInfo->pCur = NULL; pInfo->pCur = NULL;
} }
if (pInfo->pIdx) {
taosArrayDestroy(pInfo->pIdx->uids);
taosMemoryFree(pInfo->pIdx);
pInfo->pIdx = NULL;
}
taosArrayDestroy(pInfo->scanCols); taosArrayDestroy(pInfo->scanCols);
taosMemoryFreeClear(pInfo->pUser); taosMemoryFreeClear(pInfo->pUser);
@ -2790,23 +2853,619 @@ static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo,
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) { typedef int (*__optSysFilter)(void* a, void* b, int16_t dtype);
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SSysTableScanInfo* pInfo = pOperator->info; int optSysDoCompare(__compar_fn_t func, int8_t comparType, void* a, void* b) {
if (pOperator->status == OP_EXEC_DONE) { int32_t cmp = func(a, b);
return NULL; switch (comparType) {
case OP_TYPE_LOWER_THAN:
if (cmp < 0) return 0;
break;
case OP_TYPE_LOWER_EQUAL: {
if (cmp <= 0) return 0;
break;
}
case OP_TYPE_GREATER_THAN: {
if (cmp > 0) return 0;
break;
}
case OP_TYPE_GREATER_EQUAL: {
if (cmp >= 0) return 0;
break;
}
case OP_TYPE_EQUAL: {
if (cmp == 0) return 0;
break;
}
default:
return -1;
}
return 1;
} }
// the retrieve is executed on the mnode, so return tables that belongs to the information schema database. static int optSysFilterFuncImpl__LowerThan(void* a, void* b, int16_t dtype) {
if (pInfo->readHandle.mnd != NULL) { __compar_fn_t func = getComparFunc(dtype, 0);
buildSysDbTableInfo(pInfo, pOperator->resultInfo.capacity); return optSysDoCompare(func, OP_TYPE_LOWER_THAN, a, b);
}
static int optSysFilterFuncImpl__LowerEqual(void* a, void* b, int16_t dtype) {
__compar_fn_t func = getComparFunc(dtype, 0);
return optSysDoCompare(func, OP_TYPE_LOWER_EQUAL, a, b);
}
static int optSysFilterFuncImpl__GreaterThan(void* a, void* b, int16_t dtype) {
__compar_fn_t func = getComparFunc(dtype, 0);
return optSysDoCompare(func, OP_TYPE_GREATER_THAN, a, b);
}
static int optSysFilterFuncImpl__GreaterEqual(void* a, void* b, int16_t dtype) {
__compar_fn_t func = getComparFunc(dtype, 0);
return optSysDoCompare(func, OP_TYPE_GREATER_EQUAL, a, b);
}
static int optSysFilterFuncImpl__Equal(void* a, void* b, int16_t dtype) {
__compar_fn_t func = getComparFunc(dtype, 0);
return optSysDoCompare(func, OP_TYPE_EQUAL, a, b);
}
doFilterResult(pInfo); static int optSysFilterFuncImpl__NoEqual(void* a, void* b, int16_t dtype) {
pInfo->loadInfo.totalRows += pInfo->pRes->info.rows; __compar_fn_t func = getComparFunc(dtype, 0);
return optSysDoCompare(func, OP_TYPE_NOT_EQUAL, a, b);
}
static __optSysFilter optSysGetFilterFunc(int32_t ctype, bool* reverse) {
if (ctype == OP_TYPE_LOWER_EQUAL || ctype == OP_TYPE_LOWER_THAN) {
*reverse = true;
}
if (ctype == OP_TYPE_LOWER_THAN)
return optSysFilterFuncImpl__LowerThan;
else if (ctype == OP_TYPE_LOWER_EQUAL)
return optSysFilterFuncImpl__LowerEqual;
else if (ctype == OP_TYPE_GREATER_THAN)
return optSysFilterFuncImpl__GreaterThan;
else if (ctype == OP_TYPE_GREATER_EQUAL)
return optSysFilterFuncImpl__GreaterEqual;
else if (ctype == OP_TYPE_EQUAL)
return optSysFilterFuncImpl__Equal;
else if (ctype == OP_TYPE_NOT_EQUAL)
return optSysFilterFuncImpl__NoEqual;
return NULL;
}
static int32_t sysFilte__DbName(void* arg, SNode* pNode, SArray* result) {
void* pVnode = ((SSTabFltArg*)arg)->pVnode;
doSetOperatorCompleted(pOperator); const char* db = NULL;
return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes; vnodeGetInfo(pVnode, &db, NULL);
SName sn = {0};
char dbname[TSDB_DB_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&sn, varDataVal(dbname));
varDataSetLen(dbname, strlen(varDataVal(dbname)));
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
bool reverse = false;
__optSysFilter func = optSysGetFilterFunc(pOper->opType, &reverse);
if (func == NULL) return -1;
int ret = func(dbname, pVal->datum.p, TSDB_DATA_TYPE_VARCHAR);
if (ret == 0) return 0;
return -2;
}
static int32_t sysFilte__VgroupId(void* arg, SNode* pNode, SArray* result) {
void* pVnode = ((SSTabFltArg*)arg)->pVnode;
int64_t vgId = 0;
vnodeGetInfo(pVnode, NULL, (int32_t*)&vgId);
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
bool reverse = false;
__optSysFilter func = optSysGetFilterFunc(pOper->opType, &reverse);
if (func == NULL) return -1;
int ret = func(&vgId, &pVal->datum.i, TSDB_DATA_TYPE_BIGINT);
if (ret == 0) return 0;
return -1;
}
static int32_t sysFilte__TableName(void* arg, SNode* pNode, SArray* result) {
void* pMeta = ((SSTabFltArg*)arg)->pMeta;
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
bool reverse = false;
__optSysFilter func = optSysGetFilterFunc(pOper->opType, &reverse);
if (func == NULL) return -1;
SMetaFltParam param = {.suid = 0,
.cid = 0,
.type = TSDB_DATA_TYPE_VARCHAR,
.val = pVal->datum.p,
.reverse = reverse,
.filterFunc = func};
int32_t ret = metaFilterCreateTime(pMeta, &param, result);
if (ret == 0) return 0;
return -1;
}
static int32_t sysFilte__CreateTime(void* arg, SNode* pNode, SArray* result) {
void* pMeta = ((SSTabFltArg*)arg)->pMeta;
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
bool reverse = false;
__optSysFilter func = optSysGetFilterFunc(pOper->opType, &reverse);
SMetaFltParam param = {.suid = 0,
.cid = 0,
.type = TSDB_DATA_TYPE_BIGINT,
.val = &pVal->datum.i,
.reverse = reverse,
.filterFunc = func};
int32_t ret = metaFilterCreateTime(pMeta, &param, result);
if (func == NULL) return -1;
return 0;
}
static int32_t sysFilte__Ncolumn(void* arg, SNode* pNode, SArray* result) {
void* pMeta = ((SSTabFltArg*)arg)->pMeta;
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
bool reverse = false;
__optSysFilter func = optSysGetFilterFunc(pOper->opType, &reverse);
if (func == NULL) return -1;
return -1;
}
static int32_t sysFilte__Ttl(void* arg, SNode* pNode, SArray* result) {
void* pMeta = ((SSTabFltArg*)arg)->pMeta;
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
bool reverse = false;
__optSysFilter func = optSysGetFilterFunc(pOper->opType, &reverse);
if (func == NULL) return -1;
return -1;
}
static int32_t sysFilte__STableName(void* arg, SNode* pNode, SArray* result) {
void* pMeta = ((SSTabFltArg*)arg)->pMeta;
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
bool reverse = false;
__optSysFilter func = optSysGetFilterFunc(pOper->opType, &reverse);
if (func == NULL) return -1;
return -1;
}
static int32_t sysFilte__Uid(void* arg, SNode* pNode, SArray* result) {
void* pMeta = ((SSTabFltArg*)arg)->pMeta;
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
bool reverse = false;
__optSysFilter func = optSysGetFilterFunc(pOper->opType, &reverse);
if (func == NULL) return -1;
return -1;
}
static int32_t sysFilte__Type(void* arg, SNode* pNode, SArray* result) {
void* pMeta = ((SSTabFltArg*)arg)->pMeta;
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
bool reverse = false;
__optSysFilter func = optSysGetFilterFunc(pOper->opType, &reverse);
if (func == NULL) return -1;
return -1;
}
static int32_t sysChkFilter__Comm(SNode* pNode) {
// impl
SOperatorNode* pOper = (SOperatorNode*)pNode;
EOperatorType opType = pOper->opType;
if (opType != OP_TYPE_EQUAL && opType != OP_TYPE_LOWER_EQUAL && opType != OP_TYPE_LOWER_THAN &&
OP_TYPE_GREATER_EQUAL && opType != OP_TYPE_GREATER_THAN) {
return -1;
}
return 0;
}
static int32_t sysChkFilter__DBName(SNode* pNode) {
SOperatorNode* pOper = (SOperatorNode*)pNode;
if (pOper->opType != OP_TYPE_EQUAL && pOper->opType != OP_TYPE_NOT_EQUAL) {
return -1;
}
SValueNode* pVal = (SValueNode*)pOper->pRight;
if (!IS_STR_DATA_TYPE(pVal->node.resType.type)) {
return -1;
}
return 0;
}
static int32_t sysChkFilter__VgroupId(SNode* pNode) {
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
if (!IS_INTEGER_TYPE(pVal->node.resType.type)) {
return -1;
}
return sysChkFilter__Comm(pNode);
}
static int32_t sysChkFilter__TableName(SNode* pNode) {
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
if (!IS_STR_DATA_TYPE(pVal->node.resType.type)) {
return -1;
}
return sysChkFilter__Comm(pNode);
}
static int32_t sysChkFilter__CreateTime(SNode* pNode) {
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
if (!IS_TIMESTAMP_TYPE(pVal->node.resType.type)) {
return -1;
}
return sysChkFilter__Comm(pNode);
}
static int32_t sysChkFilter__Ncolumn(SNode* pNode) {
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
if (!IS_INTEGER_TYPE(pVal->node.resType.type)) {
return -1;
}
return sysChkFilter__Comm(pNode);
}
static int32_t sysChkFilter__Ttl(SNode* pNode) {
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
if (!IS_INTEGER_TYPE(pVal->node.resType.type)) {
return -1;
}
return sysChkFilter__Comm(pNode);
}
static int32_t sysChkFilter__STableName(SNode* pNode) {
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
if (!IS_STR_DATA_TYPE(pVal->node.resType.type)) {
return -1;
}
return sysChkFilter__Comm(pNode);
}
static int32_t sysChkFilter__Uid(SNode* pNode) {
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
if (!IS_INTEGER_TYPE(pVal->node.resType.type)) {
return -1;
}
return sysChkFilter__Comm(pNode);
}
static int32_t sysChkFilter__Type(SNode* pNode) {
SOperatorNode* pOper = (SOperatorNode*)pNode;
SValueNode* pVal = (SValueNode*)pOper->pRight;
if (!IS_INTEGER_TYPE(pVal->node.resType.type)) {
return -1;
}
return sysChkFilter__Comm(pNode);
}
static int32_t optSysTabFilteImpl(void* arg, SNode* cond, SArray* result) {
if (optSysCheckOper(cond) != 0) return -1;
SOperatorNode* pNode = (SOperatorNode*)cond;
int8_t i = 0;
for (; i < SYSTAB_FILTER_DICT_SIZE; i++) {
if (strcmp(filterDict[i].name, ((SColumnNode*)(pNode->pLeft))->colName) == 0) {
break;
}
}
if (i >= SYSTAB_FILTER_DICT_SIZE) return -1;
if (filterDict[i].chkFunc(cond) != 0) return -1;
return filterDict[i].fltFunc(arg, cond, result);
}
static int32_t optSysCheckOper(SNode* pOpear) {
if (nodeType(pOpear) != QUERY_NODE_OPERATOR) return -1;
SOperatorNode* pOper = (SOperatorNode*)pOpear;
if (pOper->opType < OP_TYPE_GREATER_THAN || pOper->opType > OP_TYPE_NOT_EQUAL) {
return -1;
}
if (nodeType(pOper->pLeft) != QUERY_NODE_COLUMN || nodeType(pOper->pRight) != QUERY_NODE_VALUE) {
return -1;
}
return 0;
}
static int32_t optSysMergeRslt(SArray* mRslt, SArray* rslt) {
// TODO, find comm mem from mRslt
for (int i = 0; i < taosArrayGetSize(mRslt); i++) {
SArray* aRslt = taosArrayGetP(mRslt, i);
taosArrayAddAll(rslt, aRslt);
}
return 0;
}
static int32_t optSysTabFilte(void* arg, SNode* cond, SArray* result) {
int ret = -1;
if (nodeType(cond) == QUERY_NODE_OPERATOR) {
ret = optSysTabFilteImpl(arg, cond, result);
if (ret == 0) {
SOperatorNode* pOper = (SOperatorNode*)cond;
SColumnNode* pCol = (SColumnNode*)pOper->pLeft;
if (0 == strcmp(pCol->colName, "create_time")) {
return 0;
}
return -1;
}
return ret;
}
if (nodeType(cond) != QUERY_NODE_LOGIC_CONDITION || ((SLogicConditionNode*)cond)->condType != LOGIC_COND_TYPE_AND) {
return ret;
}
SLogicConditionNode* pNode = (SLogicConditionNode*)cond;
SNodeList* pList = (SNodeList*)pNode->pParameterList;
int32_t len = LIST_LENGTH(pList);
if (len <= 0) return ret;
bool hasIdx = false;
bool hasRslt = true;
SArray* mRslt = taosArrayInit(len, POINTER_BYTES);
SListCell* cell = pList->pHead;
for (int i = 0; i < len; i++) {
if (cell == NULL) break;
SArray* aRslt = taosArrayInit(16, sizeof(int64_t));
ret = optSysTabFilteImpl(arg, cell->pNode, aRslt);
if (ret == 0) {
// has index
hasIdx = true;
taosArrayPush(mRslt, &aRslt);
} else if (ret == -2) {
// current vg
hasIdx = true;
hasRslt = false;
taosArrayDestroy(aRslt);
break;
} else { } else {
taosArrayDestroy(aRslt);
}
cell = cell->pNext;
}
if (hasRslt && hasIdx) {
optSysMergeRslt(mRslt, result);
}
for (int i = 0; i < taosArrayGetSize(mRslt); i++) {
SArray* aRslt = taosArrayGetP(mRslt, i);
taosArrayDestroy(aRslt);
}
taosArrayDestroy(mRslt);
if (hasRslt == false) {
return -2;
}
if (hasRslt && hasIdx) {
cell = pList->pHead;
for (int i = 0; i < len; i++) {
if (cell == NULL) break;
SOperatorNode* pOper = (SOperatorNode*)cell->pNode;
SColumnNode* pCol = (SColumnNode*)pOper->pLeft;
if (0 == strcmp(pCol->colName, "create_time")) {
return 0;
}
cell = cell->pNext;
}
return -1;
}
return -1;
}
static SSDataBlock* sysTableBuildUserTablesByUids(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SSysTableScanInfo* pInfo = pOperator->info;
SSysTableIndex* pIdx = pInfo->pIdx;
blockDataCleanup(pInfo->pRes);
int32_t numOfRows = 0;
int ret = 0;
const char* db = NULL;
int32_t vgId = 0;
vnodeGetInfo(pInfo->readHandle.vnode, &db, &vgId);
SName sn = {0};
char dbname[TSDB_DB_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&sn, db, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&sn, varDataVal(dbname));
varDataSetLen(dbname, strlen(varDataVal(dbname)));
SSDataBlock* p = buildInfoSchemaTableMetaBlock(TSDB_INS_TABLE_TABLES);
blockDataEnsureCapacity(p, pOperator->resultInfo.capacity);
char n[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
int32_t i = pIdx->lastIdx;
for (; i < taosArrayGetSize(pIdx->uids); i++) {
tb_uid_t* uid = taosArrayGet(pIdx->uids, i);
SMetaReader mr = {0};
metaReaderInit(&mr, pInfo->readHandle.meta, 0);
int32_t ret = metaGetTableEntryByUid(&mr, *uid);
if (ret < 0) {
metaReaderClear(&mr);
continue;
}
STR_TO_VARSTR(n, mr.me.name);
// table name
SColumnInfoData* pColInfoData = taosArrayGet(p->pDataBlock, 0);
colDataAppend(pColInfoData, numOfRows, n, false);
// database name
pColInfoData = taosArrayGet(p->pDataBlock, 1);
colDataAppend(pColInfoData, numOfRows, dbname, false);
// vgId
pColInfoData = taosArrayGet(p->pDataBlock, 6);
colDataAppend(pColInfoData, numOfRows, (char*)&vgId, false);
int32_t tableType = mr.me.type;
if (tableType == TSDB_CHILD_TABLE) {
// create time
int64_t ts = mr.me.ctbEntry.ctime;
pColInfoData = taosArrayGet(p->pDataBlock, 2);
colDataAppend(pColInfoData, numOfRows, (char*)&ts, false);
SMetaReader mr1 = {0};
metaReaderInit(&mr1, pInfo->readHandle.meta, META_READER_NOLOCK);
int64_t suid = mr.me.ctbEntry.suid;
int32_t code = metaGetTableEntryByUid(&mr1, suid);
if (code != TSDB_CODE_SUCCESS) {
qError("failed to get super table meta, cname:%s, suid:0x%" PRIx64 ", code:%s, %s", pInfo->pCur->mr.me.name,
suid, tstrerror(terrno), GET_TASKID(pTaskInfo));
metaReaderClear(&mr1);
metaReaderClear(&mr);
T_LONG_JMP(pTaskInfo->env, terrno);
}
pColInfoData = taosArrayGet(p->pDataBlock, 3);
colDataAppend(pColInfoData, numOfRows, (char*)&mr1.me.stbEntry.schemaRow.nCols, false);
// super table name
STR_TO_VARSTR(n, mr1.me.name);
pColInfoData = taosArrayGet(p->pDataBlock, 4);
colDataAppend(pColInfoData, numOfRows, n, false);
metaReaderClear(&mr1);
// table comment
pColInfoData = taosArrayGet(p->pDataBlock, 8);
if (mr.me.ctbEntry.commentLen > 0) {
char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(comment, mr.me.ctbEntry.comment);
colDataAppend(pColInfoData, numOfRows, comment, false);
} else if (mr.me.ctbEntry.commentLen == 0) {
char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(comment, "");
colDataAppend(pColInfoData, numOfRows, comment, false);
} else {
colDataAppendNULL(pColInfoData, numOfRows);
}
// uid
pColInfoData = taosArrayGet(p->pDataBlock, 5);
colDataAppend(pColInfoData, numOfRows, (char*)&mr.me.uid, false);
// ttl
pColInfoData = taosArrayGet(p->pDataBlock, 7);
colDataAppend(pColInfoData, numOfRows, (char*)&mr.me.ctbEntry.ttlDays, false);
STR_TO_VARSTR(n, "CHILD_TABLE");
} else if (tableType == TSDB_NORMAL_TABLE) {
// create time
pColInfoData = taosArrayGet(p->pDataBlock, 2);
colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.ctime, false);
// number of columns
pColInfoData = taosArrayGet(p->pDataBlock, 3);
colDataAppend(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.schemaRow.nCols, false);
// super table name
pColInfoData = taosArrayGet(p->pDataBlock, 4);
colDataAppendNULL(pColInfoData, numOfRows);
// table comment
pColInfoData = taosArrayGet(p->pDataBlock, 8);
if (mr.me.ntbEntry.commentLen > 0) {
char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(comment, mr.me.ntbEntry.comment);
colDataAppend(pColInfoData, numOfRows, comment, false);
} else if (mr.me.ntbEntry.commentLen == 0) {
char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(comment, "");
colDataAppend(pColInfoData, numOfRows, comment, false);
} else {
colDataAppendNULL(pColInfoData, numOfRows);
}
// uid
pColInfoData = taosArrayGet(p->pDataBlock, 5);
colDataAppend(pColInfoData, numOfRows, (char*)&mr.me.uid, false);
// ttl
pColInfoData = taosArrayGet(p->pDataBlock, 7);
colDataAppend(pColInfoData, numOfRows, (char*)&mr.me.ntbEntry.ttlDays, false);
STR_TO_VARSTR(n, "NORMAL_TABLE");
// impl later
}
metaReaderClear(&mr);
pColInfoData = taosArrayGet(p->pDataBlock, 9);
colDataAppend(pColInfoData, numOfRows, n, false);
if (++numOfRows >= pOperator->resultInfo.capacity) {
p->info.rows = numOfRows;
pInfo->pRes->info.rows = numOfRows;
relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
doFilterResult(pInfo);
blockDataCleanup(p);
numOfRows = 0;
if (pInfo->pRes->info.rows > 0) {
break;
}
}
}
if (numOfRows > 0) {
p->info.rows = numOfRows;
pInfo->pRes->info.rows = numOfRows;
relocateColumnData(pInfo->pRes, pInfo->scanCols, p->pDataBlock, false);
doFilterResult(pInfo);
blockDataCleanup(p);
numOfRows = 0;
}
if (i >= taosArrayGetSize(pIdx->uids)) {
doSetOperatorCompleted(pOperator);
} else {
pIdx->lastIdx = i;
}
blockDataDestroy(p);
pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
}
static SSDataBlock* sysTableBuildUserTables(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SSysTableScanInfo* pInfo = pOperator->info;
if (pInfo->pCur == NULL) { if (pInfo->pCur == NULL) {
pInfo->pCur = metaOpenTbCursor(pInfo->readHandle.meta); pInfo->pCur = metaOpenTbCursor(pInfo->readHandle.meta);
} }
@ -2980,6 +3639,58 @@ static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) {
pInfo->loadInfo.totalRows += pInfo->pRes->info.rows; pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes; return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
} }
static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SSysTableScanInfo* pInfo = pOperator->info;
SNode* pCondition = pInfo->pCondition;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
}
// the retrieve is executed on the mnode, so return tables that belongs to the information schema database.
if (pInfo->readHandle.mnd != NULL) {
buildSysDbTableInfo(pInfo, pOperator->resultInfo.capacity);
doFilterResult(pInfo);
pInfo->loadInfo.totalRows += pInfo->pRes->info.rows;
doSetOperatorCompleted(pOperator);
return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes;
} else {
if (pInfo->showRewrite == false) {
if (pCondition != NULL && pInfo->pIdx == NULL) {
SSTabFltArg arg = {.pMeta = pInfo->readHandle.meta, .pVnode = pInfo->readHandle.vnode};
SSysTableIndex* idx = taosMemoryMalloc(sizeof(SSysTableIndex));
idx->init = 0;
idx->uids = taosArrayInit(128, sizeof(int64_t));
idx->lastIdx = 0;
pInfo->pIdx = idx; // set idx arg
int flt = optSysTabFilte(&arg, pCondition, idx->uids);
if (flt == 0) {
pInfo->pIdx->init = 1;
SSDataBlock* blk = sysTableBuildUserTablesByUids(pOperator);
return blk;
} else if (flt == -2) {
qDebug("%s failed to get sys table info by idx, empty result", GET_TASKID(pTaskInfo));
return NULL;
} else if (flt == -1) {
// not idx
qDebug("%s failed to get sys table info by idx, scan sys table one by one", GET_TASKID(pTaskInfo));
}
} else if (pCondition != NULL && (pInfo->pIdx != NULL && pInfo->pIdx->init == 1)) {
SSDataBlock* blk = sysTableBuildUserTablesByUids(pOperator);
return blk;
}
}
return sysTableBuildUserTables(pOperator);
}
return NULL;
} }
static SSDataBlock* sysTableScanUserSTables(SOperatorInfo* pOperator) { static SSDataBlock* sysTableScanUserSTables(SOperatorInfo* pOperator) {

12
source/libs/index/inc/indexFst.h
View File

@ -290,21 +290,21 @@ bool fstVerify(Fst* fst);
// refactor this function // refactor this function
bool fstBuilderNodeCompileTo(FstBuilderNode* b, IdxFstFile* wrt, CompiledAddr lastAddr, CompiledAddr startAddr); bool fstBuilderNodeCompileTo(FstBuilderNode* b, IdxFstFile* wrt, CompiledAddr lastAddr, CompiledAddr startAddr);
typedef struct StreamState { typedef struct FstStreamState {
FstNode* node; FstNode* node;
uint64_t trans; uint64_t trans;
FstOutput out; FstOutput out;
void* autState; void* autState;
} StreamState; } FstStreamState;
void streamStateDestroy(void* s); void fstStreamStateDestroy(void* s);
typedef struct FStmSt { typedef struct FStmSt {
Fst* fst; Fst* fst;
FAutoCtx* aut; FAutoCtx* aut;
SArray* inp; SArray* inp;
FstOutput emptyOutput; FstOutput emptyOutput;
SArray* stack; // <StreamState> SArray* stack; // <FstStreamState>
FstBoundWithData* endAt; FstBoundWithData* endAt;
} FStmSt; } FStmSt;
@ -317,14 +317,14 @@ typedef struct FStmStRslt {
FStmStRslt* swsResultCreate(FstSlice* data, FstOutput fOut, void* state); FStmStRslt* swsResultCreate(FstSlice* data, FstOutput fOut, void* state);
void swsResultDestroy(FStmStRslt* result); void swsResultDestroy(FStmStRslt* result);
typedef void* (*StreamCallback)(void*); typedef void* (*streamCallback__fn)(void*);
FStmSt* stmStCreate(Fst* fst, FAutoCtx* automation, FstBoundWithData* min, FstBoundWithData* max); FStmSt* stmStCreate(Fst* fst, FAutoCtx* automation, FstBoundWithData* min, FstBoundWithData* max);
void stmStDestroy(FStmSt* sws); void stmStDestroy(FStmSt* sws);
bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min); bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min);
FStmStRslt* stmStNextWith(FStmSt* sws, StreamCallback callback); FStmStRslt* stmStNextWith(FStmSt* sws, streamCallback__fn callback);
FStmBuilder* stmBuilderCreate(Fst* fst, FAutoCtx* aut); FStmBuilder* stmBuilderCreate(Fst* fst, FAutoCtx* aut);

36
source/libs/index/src/indexFst.c
View File

@ -1165,7 +1165,7 @@ FStmSt* stmStCreate(Fst* fst, FAutoCtx* automation, FstBoundWithData* min, FstBo
sws->emptyOutput.null = true; sws->emptyOutput.null = true;
sws->emptyOutput.out = 0; sws->emptyOutput.out = 0;
sws->stack = (SArray*)taosArrayInit(256, sizeof(StreamState)); sws->stack = (SArray*)taosArrayInit(256, sizeof(FstStreamState));
sws->endAt = max; sws->endAt = max;
stmStSeekMin(sws, min); stmStSeekMin(sws, min);
@ -1177,7 +1177,7 @@ void stmStDestroy(FStmSt* sws) {
} }
taosArrayDestroy(sws->inp); taosArrayDestroy(sws->inp);
taosArrayDestroyEx(sws->stack, streamStateDestroy); taosArrayDestroyEx(sws->stack, fstStreamStateDestroy);
taosMemoryFree(sws); taosMemoryFree(sws);
} }
@ -1188,7 +1188,7 @@ bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min) {
if (fstBoundWithDataIsIncluded(min)) { if (fstBoundWithDataIsIncluded(min)) {
sws->emptyOutput.out = fstEmptyFinalOutput(sws->fst, &(sws->emptyOutput.null)); sws->emptyOutput.out = fstEmptyFinalOutput(sws->fst, &(sws->emptyOutput.null));
} }
StreamState s = {.node = fstGetRoot(sws->fst), FstStreamState s = {.node = fstGetRoot(sws->fst),
.trans = 0, .trans = 0,
.out = {.null = false, .out = 0}, .out = {.null = false, .out = 0},
.autState = automFuncs[aut->type].start(aut)}; // auto.start callback .autState = automFuncs[aut->type].start(aut)}; // auto.start callback
@ -1223,7 +1223,7 @@ bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min) {
autState = automFuncs[aut->type].accept(aut, preState, b); autState = automFuncs[aut->type].accept(aut, preState, b);
taosArrayPush(sws->inp, &b); taosArrayPush(sws->inp, &b);
StreamState s = {.node = node, .trans = res + 1, .out = {.null = false, .out = out}, .autState = preState}; FstStreamState s = {.node = node, .trans = res + 1, .out = {.null = false, .out = out}, .autState = preState};
node = NULL; node = NULL;
taosArrayPush(sws->stack, &s); taosArrayPush(sws->stack, &s);
@ -1244,7 +1244,7 @@ bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min) {
} }
} }
StreamState s = {.node = node, .trans = i, .out = {.null = false, .out = out}, .autState = autState}; FstStreamState s = {.node = node, .trans = i, .out = {.null = false, .out = out}, .autState = autState};
taosArrayPush(sws->stack, &s); taosArrayPush(sws->stack, &s);
taosMemoryFree(trans); taosMemoryFree(trans);
return true; return true;
@ -1255,7 +1255,7 @@ bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min) {
uint32_t sz = taosArrayGetSize(sws->stack); uint32_t sz = taosArrayGetSize(sws->stack);
if (sz != 0) { if (sz != 0) {
StreamState* s = taosArrayGet(sws->stack, sz - 1); FstStreamState* s = taosArrayGet(sws->stack, sz - 1);
if (inclusize) { if (inclusize) {
s->trans -= 1; s->trans -= 1;
taosArrayPop(sws->inp); taosArrayPop(sws->inp);
@ -1264,7 +1264,7 @@ bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min) {
uint64_t trans = s->trans; uint64_t trans = s->trans;
FstTransition trn; FstTransition trn;
fstNodeGetTransitionAt(n, trans - 1, &trn); fstNodeGetTransitionAt(n, trans - 1, &trn);
StreamState s = { FstStreamState s = {
.node = fstGetNode(sws->fst, trn.addr), .trans = 0, .out = {.null = false, .out = out}, .autState = autState}; .node = fstGetNode(sws->fst, trn.addr), .trans = 0, .out = {.null = false, .out = out}, .autState = autState};
taosArrayPush(sws->stack, &s); taosArrayPush(sws->stack, &s);
return true; return true;
@ -1274,14 +1274,14 @@ bool stmStSeekMin(FStmSt* sws, FstBoundWithData* min) {
return false; return false;
} }
FStmStRslt* stmStNextWith(FStmSt* sws, StreamCallback callback) { FStmStRslt* stmStNextWith(FStmSt* sws, streamCallback__fn callback) {
FAutoCtx* aut = sws->aut; FAutoCtx* aut = sws->aut;
FstOutput output = sws->emptyOutput; FstOutput output = sws->emptyOutput;
if (output.null == false) { if (output.null == false) {
FstSlice emptySlice = fstSliceCreate(NULL, 0); FstSlice emptySlice = fstSliceCreate(NULL, 0);
if (fstBoundWithDataExceededBy(sws->endAt, &emptySlice)) { if (fstBoundWithDataExceededBy(sws->endAt, &emptySlice)) {
taosArrayDestroyEx(sws->stack, streamStateDestroy); taosArrayDestroyEx(sws->stack, fstStreamStateDestroy);
sws->stack = (SArray*)taosArrayInit(256, sizeof(StreamState)); sws->stack = (SArray*)taosArrayInit(256, sizeof(FstStreamState));
return NULL; return NULL;
} }
void* start = automFuncs[aut->type].start(aut); void* start = automFuncs[aut->type].start(aut);
@ -1292,12 +1292,12 @@ FStmStRslt* stmStNextWith(FStmSt* sws, StreamCallback callback) {
} }
SArray* nodes = taosArrayInit(8, sizeof(FstNode*)); SArray* nodes = taosArrayInit(8, sizeof(FstNode*));
while (taosArrayGetSize(sws->stack) > 0) { while (taosArrayGetSize(sws->stack) > 0) {
StreamState* p = (StreamState*)taosArrayPop(sws->stack); FstStreamState* p = (FstStreamState*)taosArrayPop(sws->stack);
if (p->trans >= FST_NODE_LEN(p->node) || !automFuncs[aut->type].canMatch(aut, p->autState)) { if (p->trans >= FST_NODE_LEN(p->node) || !automFuncs[aut->type].canMatch(aut, p->autState)) {
if (FST_NODE_ADDR(p->node) != fstGetRootAddr(sws->fst)) { if (FST_NODE_ADDR(p->node) != fstGetRootAddr(sws->fst)) {
taosArrayPop(sws->inp); taosArrayPop(sws->inp);
} }
streamStateDestroy(p); fstStreamStateDestroy(p);
continue; continue;
} }
FstTransition trn; FstTransition trn;
@ -1318,10 +1318,10 @@ FStmStRslt* stmStNextWith(FStmSt* sws, StreamCallback callback) {
isMatch = automFuncs[aut->type].isMatch(aut, eofState); isMatch = automFuncs[aut->type].isMatch(aut, eofState);
} }
} }
StreamState s1 = {.node = p->node, .trans = p->trans + 1, .out = p->out, .autState = p->autState}; FstStreamState s1 = {.node = p->node, .trans = p->trans + 1, .out = p->out, .autState = p->autState};
taosArrayPush(sws->stack, &s1); taosArrayPush(sws->stack, &s1);
StreamState s2 = {.node = nextNode, .trans = 0, .out = {.null = false, .out = out}, .autState = nextState}; FstStreamState s2 = {.node = nextNode, .trans = 0, .out = {.null = false, .out = out}, .autState = nextState};
taosArrayPush(sws->stack, &s2); taosArrayPush(sws->stack, &s2);
int32_t isz = taosArrayGetSize(sws->inp); int32_t isz = taosArrayGetSize(sws->inp);
@ -1331,8 +1331,8 @@ FStmStRslt* stmStNextWith(FStmSt* sws, StreamCallback callback) {
} }
FstSlice slice = fstSliceCreate(buf, isz); FstSlice slice = fstSliceCreate(buf, isz);
if (fstBoundWithDataExceededBy(sws->endAt, &slice)) { if (fstBoundWithDataExceededBy(sws->endAt, &slice)) {
taosArrayDestroyEx(sws->stack, streamStateDestroy); taosArrayDestroyEx(sws->stack, fstStreamStateDestroy);
sws->stack = (SArray*)taosArrayInit(256, sizeof(StreamState)); sws->stack = (SArray*)taosArrayInit(256, sizeof(FstStreamState));
taosMemoryFreeClear(buf); taosMemoryFreeClear(buf);
fstSliceDestroy(&slice); fstSliceDestroy(&slice);
taosArrayDestroy(nodes); taosArrayDestroy(nodes);
@ -1375,11 +1375,11 @@ void swsResultDestroy(FStmStRslt* result) {
taosMemoryFree(result); taosMemoryFree(result);
} }
void streamStateDestroy(void* s) { void fstStreamStateDestroy(void* s) {
if (NULL == s) { if (NULL == s) {
return; return;
} }
StreamState* ss = (StreamState*)s; FstStreamState* ss = (FstStreamState*)s;
fstNodeDestroy(ss->node); fstNodeDestroy(ss->node);
} }

2
source/libs/index/src/indexJson.c
View File

@ -30,7 +30,7 @@ int indexJsonPut(SIndexJson *index, SIndexJsonMultiTerm *terms, uint64_t uid) {
} else { } else {
p->colType = TSDB_DATA_TYPE_DOUBLE; p->colType = TSDB_DATA_TYPE_DOUBLE;
} }
IDX_TYPE_ADD_EXTERN_TYPE(p->colType, TSDB_DATA_TYPE_JSON); IDX_TYPE_ADD_EXTERN_TYPE((p->colType), TSDB_DATA_TYPE_JSON);
} }
// handle put // handle put
return indexPut(index, terms, uid); return indexPut(index, terms, uid);

1
source/libs/transport/src/trans.c
View File

@ -69,6 +69,7 @@ void* rpcOpen(const SRpcInit* pInit) {
pRpc->idleTime = pInit->idleTime; pRpc->idleTime = pInit->idleTime;
pRpc->tcphandle = pRpc->tcphandle =
(*taosInitHandle[pRpc->connType])(ip, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc); (*taosInitHandle[pRpc->connType])(ip, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);
if (pRpc->tcphandle == NULL) { if (pRpc->tcphandle == NULL) {
taosMemoryFree(pRpc); taosMemoryFree(pRpc);
return NULL; return NULL;

9
source/libs/transport/src/transCli.c
View File

@ -439,12 +439,14 @@ void cliHandleExceptImpl(SCliConn* pConn, int32_t code) {
tDebug("%s conn %p construct ahandle %p by %s", CONN_GET_INST_LABEL(pConn), pConn, transMsg.info.ahandle, tDebug("%s conn %p construct ahandle %p by %s", CONN_GET_INST_LABEL(pConn), pConn, transMsg.info.ahandle,
TMSG_INFO(transMsg.msgType)); TMSG_INFO(transMsg.msgType));
if (transMsg.info.ahandle == NULL) { if (transMsg.info.ahandle == NULL) {
transMsg.info.ahandle = transCtxDumpBrokenlinkVal(&pConn->ctx, (int32_t*)&(transMsg.msgType)); int32_t msgType = 0;
transMsg.info.ahandle = transCtxDumpBrokenlinkVal(&pConn->ctx, &msgType);
transMsg.msgType = msgType;
tDebug("%s conn %p construct ahandle %p due to brokenlink", CONN_GET_INST_LABEL(pConn), pConn, tDebug("%s conn %p construct ahandle %p due to brokenlink", CONN_GET_INST_LABEL(pConn), pConn,
transMsg.info.ahandle); transMsg.info.ahandle);
} }
} else { } else {
transMsg.info.ahandle = (pMsg->type != Release && pCtx) ? pCtx->ahandle : NULL; transMsg.info.ahandle = (pMsg != NULL && pMsg->type != Release && pCtx) ? pCtx->ahandle : NULL;
} }
if (pCtx == NULL || pCtx->pSem == NULL) { if (pCtx == NULL || pCtx->pSem == NULL) {
@ -1078,9 +1080,6 @@ static void cliPrepareCb(uv_prepare_t* handle) {
QUEUE_REMOVE(h); QUEUE_REMOVE(h);
SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q); SCliMsg* pMsg = QUEUE_DATA(h, SCliMsg, q);
if (pMsg == NULL) {
continue;
}
(*cliAsyncHandle[pMsg->type])(pMsg, thrd); (*cliAsyncHandle[pMsg->type])(pMsg, thrd);
count++; count++;
} }

107
source/os/src/osSocket.c
View File

@ -73,6 +73,7 @@ typedef struct TdEpoll {
EpollFd fd; EpollFd fd;
} * TdEpollPtr, TdEpoll; } * TdEpollPtr, TdEpoll;
#if 0
int32_t taosSendto(TdSocketPtr pSocket, void *buf, int len, unsigned int flags, const struct sockaddr *dest_addr, int32_t taosSendto(TdSocketPtr pSocket, void *buf, int len, unsigned int flags, const struct sockaddr *dest_addr,
int addrlen) { int addrlen) {
if (pSocket == NULL || pSocket->fd < 0) { if (pSocket == NULL || pSocket->fd < 0) {
@ -84,6 +85,7 @@ int32_t taosSendto(TdSocketPtr pSocket, void *buf, int len, unsigned int flags,
return sendto(pSocket->fd, buf, len, flags, dest_addr, addrlen); return sendto(pSocket->fd, buf, len, flags, dest_addr, addrlen);
#endif #endif
} }
int32_t taosWriteSocket(TdSocketPtr pSocket, void *buf, int len) { int32_t taosWriteSocket(TdSocketPtr pSocket, void *buf, int len) {
if (pSocket == NULL || pSocket->fd < 0) { if (pSocket == NULL || pSocket->fd < 0) {
return -1; return -1;
@ -114,6 +116,8 @@ int32_t taosReadFromSocket(TdSocketPtr pSocket, void *buf, int32_t len, int32_t
} }
return recvfrom(pSocket->fd, buf, len, flags, destAddr, addrLen); return recvfrom(pSocket->fd, buf, len, flags, destAddr, addrLen);
} }
#endif // endif 0
int32_t taosCloseSocketNoCheck1(SocketFd fd) { int32_t taosCloseSocketNoCheck1(SocketFd fd) {
#ifdef WINDOWS #ifdef WINDOWS
return closesocket(fd); return closesocket(fd);
@ -121,6 +125,7 @@ int32_t taosCloseSocketNoCheck1(SocketFd fd) {
return close(fd); return close(fd);
#endif #endif
} }
int32_t taosCloseSocket(TdSocketPtr *ppSocket) { int32_t taosCloseSocket(TdSocketPtr *ppSocket) {
int32_t code; int32_t code;
if (ppSocket == NULL || *ppSocket == NULL || (*ppSocket)->fd < 0) { if (ppSocket == NULL || *ppSocket == NULL || (*ppSocket)->fd < 0) {
@ -131,6 +136,8 @@ int32_t taosCloseSocket(TdSocketPtr *ppSocket) {
taosMemoryFree(*ppSocket); taosMemoryFree(*ppSocket);
return code; return code;
} }
#if 0
int32_t taosCloseSocketServer(TdSocketServerPtr *ppSocketServer) { int32_t taosCloseSocketServer(TdSocketServerPtr *ppSocketServer) {
int32_t code; int32_t code;
if (ppSocketServer == NULL || *ppSocketServer == NULL || (*ppSocketServer)->fd < 0) { if (ppSocketServer == NULL || *ppSocketServer == NULL || (*ppSocketServer)->fd < 0) {
@ -216,20 +223,6 @@ int32_t taosShutDownSocketServerRDWR(TdSocketServerPtr pSocketServer) {
#endif #endif
} }
void taosWinSocketInit() {
#ifdef WINDOWS
static char flag = 0;
if (flag == 0) {
WORD wVersionRequested;
WSADATA wsaData;
wVersionRequested = MAKEWORD(1, 1);
if (WSAStartup(wVersionRequested, &wsaData) == 0) {
flag = 1;
}
}
#else
#endif
}
int32_t taosSetNonblocking(TdSocketPtr pSocket, int32_t on) { int32_t taosSetNonblocking(TdSocketPtr pSocket, int32_t on) {
if (pSocket == NULL || pSocket->fd < 0) { if (pSocket == NULL || pSocket->fd < 0) {
return -1; return -1;
@ -262,6 +255,8 @@ int32_t taosSetNonblocking(TdSocketPtr pSocket, int32_t on) {
#endif #endif
return 0; return 0;
} }
#endif // endif 0
int32_t taosSetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void *optval, int32_t optlen) { int32_t taosSetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void *optval, int32_t optlen) {
if (pSocket == NULL || pSocket->fd < 0) { if (pSocket == NULL || pSocket->fd < 0) {
return -1; return -1;
@ -296,6 +291,8 @@ int32_t taosSetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void
return setsockopt(pSocket->fd, level, optname, optval, (int)optlen); return setsockopt(pSocket->fd, level, optname, optval, (int)optlen);
#endif #endif
} }
#if 0
int32_t taosGetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void *optval, int32_t *optlen) { int32_t taosGetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void *optval, int32_t *optlen) {
if (pSocket == NULL || pSocket->fd < 0) { if (pSocket == NULL || pSocket->fd < 0) {
return -1; return -1;
@ -307,6 +304,9 @@ int32_t taosGetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void
return getsockopt(pSocket->fd, level, optname, optval, (int *)optlen); return getsockopt(pSocket->fd, level, optname, optval, (int *)optlen);
#endif #endif
} }
#endif
uint32_t taosInetAddr(const char *ipAddr) { uint32_t taosInetAddr(const char *ipAddr) {
#ifdef WINDOWS #ifdef WINDOWS
uint32_t value; uint32_t value;
@ -330,6 +330,7 @@ const char *taosInetNtoa(struct in_addr ipInt, char *dstStr, int32_t len) {
#define TCP_CONN_TIMEOUT 3000 // conn timeout #define TCP_CONN_TIMEOUT 3000 // conn timeout
#if 0
int32_t taosWriteMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) { int32_t taosWriteMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) {
if (pSocket == NULL || pSocket->fd < 0) { if (pSocket == NULL || pSocket->fd < 0) {
return -1; return -1;
@ -726,6 +727,7 @@ int taosValidIp(uint32_t ip) {
#endif #endif
return 0; return 0;
} }
#endif // endif 0
bool taosValidIpAndPort(uint32_t ip, uint16_t port) { bool taosValidIpAndPort(uint32_t ip, uint16_t port) {
struct sockaddr_in serverAdd; struct sockaddr_in serverAdd;
@ -774,6 +776,8 @@ bool taosValidIpAndPort(uint32_t ip, uint16_t port) {
return true; return true;
// return 0 == taosValidIp(ip) ? true : false; // return 0 == taosValidIp(ip) ? true : false;
} }
#if 0
TdSocketServerPtr taosOpenTcpServerSocket(uint32_t ip, uint16_t port) { TdSocketServerPtr taosOpenTcpServerSocket(uint32_t ip, uint16_t port) {
struct sockaddr_in serverAdd; struct sockaddr_in serverAdd;
SocketFd fd; SocketFd fd;
@ -888,6 +892,36 @@ int64_t taosCopyFds(TdSocketPtr pSrcSocket, TdSocketPtr pDestSocket, int64_t len
return len; return len;
} }
// Function converting an IP address string to an uint32_t.
uint32_t ip2uint(const char *const ip_addr) {
char ip_addr_cpy[20];
char ip[5];
tstrncpy(ip_addr_cpy, ip_addr, sizeof(ip_addr_cpy));
char *s_start, *s_end;
s_start = ip_addr_cpy;
s_end = ip_addr_cpy;
int32_t k;
for (k = 0; *s_start != '\0'; s_start = s_end) {
for (s_end = s_start; *s_end != '.' && *s_end != '\0'; s_end++) {
}
if (*s_end == '.') {
*s_end = '\0';
s_end++;
}
ip[k++] = (char)atoi(s_start);
}
ip[k] = '\0';
return *((uint32_t *)ip);
}
#endif // endif 0
void taosBlockSIGPIPE() { void taosBlockSIGPIPE() {
#ifdef WINDOWS #ifdef WINDOWS
// assert(0); // assert(0);
@ -991,34 +1025,6 @@ int32_t taosGetFqdn(char *fqdn) {
return 0; return 0;
} }
// Function converting an IP address string to an uint32_t.
uint32_t ip2uint(const char *const ip_addr) {
char ip_addr_cpy[20];
char ip[5];
tstrncpy(ip_addr_cpy, ip_addr, sizeof(ip_addr_cpy));
char *s_start, *s_end;
s_start = ip_addr_cpy;
s_end = ip_addr_cpy;
int32_t k;
for (k = 0; *s_start != '\0'; s_start = s_end) {
for (s_end = s_start; *s_end != '.' && *s_end != '\0'; s_end++) {
}
if (*s_end == '.') {
*s_end = '\0';
s_end++;
}
ip[k++] = (char)atoi(s_start);
}
ip[k] = '\0';
return *((uint32_t *)ip);
}
void tinet_ntoa(char *ipstr, uint32_t ip) { void tinet_ntoa(char *ipstr, uint32_t ip) {
sprintf(ipstr, "%d.%d.%d.%d", ip & 0xFF, (ip >> 8) & 0xFF, (ip >> 16) & 0xFF, ip >> 24); sprintf(ipstr, "%d.%d.%d.%d", ip & 0xFF, (ip >> 8) & 0xFF, (ip >> 16) & 0xFF, ip >> 24);
} }
@ -1039,12 +1045,14 @@ void taosSetMaskSIGPIPE() {
#endif #endif
} }
#if 0
int32_t taosGetSocketName(TdSocketPtr pSocket, struct sockaddr *destAddr, int *addrLen) { int32_t taosGetSocketName(TdSocketPtr pSocket, struct sockaddr *destAddr, int *addrLen) {
if (pSocket == NULL || pSocket->fd < 0) { if (pSocket == NULL || pSocket->fd < 0) {
return -1; return -1;
} }
return getsockname(pSocket->fd, destAddr, addrLen); return getsockname(pSocket->fd, destAddr, addrLen);
} }
#endif // endif 0
/* /*
* Set TCP connection timeout per-socket level. * Set TCP connection timeout per-socket level.
@ -1080,3 +1088,18 @@ int32_t taosCreateSocketWithTimeout(uint32_t timeout) {
return (int)fd; return (int)fd;
} }
void taosWinSocketInit() {
#ifdef WINDOWS
static char flag = 0;
if (flag == 0) {
WORD wVersionRequested;
WSADATA wsaData;
wVersionRequested = MAKEWORD(1, 1);
if (WSAStartup(wVersionRequested, &wsaData) == 0) {
flag = 1;
}
}
#else
#endif
}

22
tests/script/tsim/parser/where.sim
View File

@ -87,17 +87,17 @@ if $rows != 2 then
return -1 return -1
endi endi
print $tbPrefix #print $tbPrefix
$tb = $tbPrefix . 0 #$tb = $tbPrefix . 0
if $data00 != wh_tb1 then #if $data00 != wh_tb1 then
print expect wh_tb1, actual:$data00 # print expect wh_tb1, actual:$data00
return -1 # return -1
endi #endi
$tb = $tbPrefix . 1 #$tb = $tbPrefix . 1
if $data10 != wh_tb0 then #if $data10 != wh_tb0 then
print expect wh_tb0, actual:$data00 # print expect wh_tb0, actual:$data00
return -1 # return -1
endi #endi
## select specified columns ## select specified columns