enh: add table blokc order check
This commit is contained in:
parent
932116e4b0
commit
9e39ffc848
|
@ -155,6 +155,7 @@ typedef struct STableDataCxt {
|
||||||
SSubmitTbData *pData;
|
SSubmitTbData *pData;
|
||||||
TSKEY lastTs;
|
TSKEY lastTs;
|
||||||
bool ordered;
|
bool ordered;
|
||||||
|
bool duplicateTs;
|
||||||
} STableDataCxt;
|
} STableDataCxt;
|
||||||
|
|
||||||
typedef struct SVgroupDataCxt {
|
typedef struct SVgroupDataCxt {
|
||||||
|
|
|
@ -982,15 +982,19 @@ int32_t insInitBoundColsInfo(int32_t numOfBound, SBoundColInfo* pInfo) {
|
||||||
}
|
}
|
||||||
|
|
||||||
void insCheckTableDataOrder(STableDataCxt* pTableCxt, TSKEY tsKey) {
|
void insCheckTableDataOrder(STableDataCxt* pTableCxt, TSKEY tsKey) {
|
||||||
// once the data block is disordered, we do NOT keep previous timestamp any more
|
// once the data block is disordered, we do NOT keep last timestamp any more
|
||||||
if (!pTableCxt->ordered) {
|
if (!pTableCxt->ordered) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (tsKey <= pTableCxt->lastTs) {
|
if (tsKey < pTableCxt->lastTs) {
|
||||||
pTableCxt->ordered = false;
|
pTableCxt->ordered = false;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (tsKey == pTableCxt->lastTs) {
|
||||||
|
pTableCxt->duplicateTs = true;
|
||||||
|
}
|
||||||
|
|
||||||
pTableCxt->lastTs = tsKey;
|
pTableCxt->lastTs = tsKey;
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
@ -1005,8 +1009,9 @@ static int32_t createTableDataCxt(STableMeta* pTableMeta, SVCreateTbReq** pCreat
|
||||||
|
|
||||||
int32_t code = TSDB_CODE_SUCCESS;
|
int32_t code = TSDB_CODE_SUCCESS;
|
||||||
|
|
||||||
pTableCxt->ordered = true;
|
|
||||||
pTableCxt->lastTs = 0;
|
pTableCxt->lastTs = 0;
|
||||||
|
pTableCxt->ordered = true;
|
||||||
|
pTableCxt->duplicateTs = false;
|
||||||
|
|
||||||
pTableCxt->pMeta = tableMetaDup(pTableMeta);
|
pTableCxt->pMeta = tableMetaDup(pTableMeta);
|
||||||
if (NULL == pTableCxt->pMeta) {
|
if (NULL == pTableCxt->pMeta) {
|
||||||
|
@ -1196,10 +1201,12 @@ int32_t insMergeTableDataCxt(SHashObj* pTableHash, SArray** pVgDataBlocks) {
|
||||||
void* p = taosHashIterate(pTableHash, NULL);
|
void* p = taosHashIterate(pTableHash, NULL);
|
||||||
while (TSDB_CODE_SUCCESS == code && NULL != p) {
|
while (TSDB_CODE_SUCCESS == code && NULL != p) {
|
||||||
STableDataCxt* pTableCxt = *(STableDataCxt**)p;
|
STableDataCxt* pTableCxt = *(STableDataCxt**)p;
|
||||||
if (pTableCxt->ordered) {
|
if (!pTableCxt->ordered) {
|
||||||
tRowSort(pTableCxt->pData->aRowP);
|
tRowSort(pTableCxt->pData->aRowP);
|
||||||
}
|
}
|
||||||
code = tRowMerge(pTableCxt->pData->aRowP, pTableCxt->pSchema, 0);
|
if (!pTableCxt->ordered || pTableCxt->duplicateTs) {
|
||||||
|
code = tRowMerge(pTableCxt->pData->aRowP, pTableCxt->pSchema, 0);
|
||||||
|
}
|
||||||
if (TSDB_CODE_SUCCESS == code) {
|
if (TSDB_CODE_SUCCESS == code) {
|
||||||
SVgroupDataCxt* pVgCxt = NULL;
|
SVgroupDataCxt* pVgCxt = NULL;
|
||||||
int32_t vgId = pTableCxt->pMeta->vgId;
|
int32_t vgId = pTableCxt->pMeta->vgId;
|
||||||
|
|
Loading…
Reference in New Issue