#4548

Committed 22 Jul 2025 02:37AM UTC coverage: 54.273% (-3.0%) from 57.287%

Build # #4548

Build Type

push

travis-ci

Committed by GitHub

Commit Message

Merge pull request #32061 from taosdata/new_testcases

Run Details

132738 of 315239 branches covered (42.11%)

Branch coverage included in aggregate %.

201371 of 300373 relevant lines covered (67.04%)

3475977.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

8.66

/source/dnode/vnode/src/bse/bseTableMgt.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "bseTableMgt.h"
#include "bse.h"
#include "bseTable.h"
#include "bseUtil.h"
#include "thash.h"

static int32_t tableReaderMgtInit(STableReaderMgt *pReader, SBse *pBse, int64_t retention);
static void    tableReaderMgtSetRetion(STableReaderMgt *pReader, int64_t retention);
static int32_t tableReaderMgtSeek(STableReaderMgt *pReaderMgt, int64_t seq, uint8_t **pValue, int32_t *len);
static int32_t tableReaderMgtClear(STableReaderMgt *pReader);
static void    tableReaderMgtDestroy(STableReaderMgt *pReader);

static int32_t tableBuilderMgtInit(STableBuilderMgt *pMgt, SBse *pBse, int64_t retention);
static void    tableBuilderMgtSetRetion(STableBuilderMgt *pMgt, int64_t retention);
static int32_t tableBuilderMgtGetBuilder(STableBuilderMgt *pMgt, int64_t seq, STableBuilder **p);
static int32_t tableBuilderMgtCommit(STableBuilderMgt *pMgt, SBseLiveFileInfo *pInfo);
static int32_t tableBuilderMgtSeek(STableBuilderMgt *pMgt, int64_t seq, uint8_t **pValue, int32_t *len);
static int32_t tableBuilderMgtPutBatch(STableBuilderMgt *pMgt, SBseBatch *pBatch);
static int32_t tableBuilderMgtClear(STableBuilderMgt *pMgt);
static void    tableBuilderMgtDestroy(STableBuilderMgt *pMgt);

static int32_t tableBuilderMgtRecoverTable(STableBuilderMgt *pMgt, int64_t seq, STableBuilder **pBuilder, int64_t size);

static int32_t tableMetaMgtInit(STableMetaMgt *pMgt, SBse *pBse, int64_t retention);
static void    tableMetaMgtDestroy(STableMetaMgt *pMgt);

static void tableReaderFree(void *pReader);

static void blockFree(void *pBlock);

int32_t bseTableMgtCreate(SBse *pBse, void **pMgt) {
  int32_t code = 0;
  int32_t lino = 0;

  STableMgt *p = taosMemoryCalloc(1, sizeof(STableMgt));
  if (p == NULL) {
    return terrno;
  }
  p->pBse = pBse;
  p->pHashObj = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
  if (p->pHashObj == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _error);
  }

  *pMgt = p;
_error:
  if (code != 0) {
    if (p != NULL)
      bseError("vgId:%d failed to open table pBuilderMgt at line %d since %s", BSE_GET_VGID((SBse *)p->pBse), lino,
               tstrerror(code));
    bseTableMgtCleanup(p);
  }
  return code;
}

int32_t bseTableMgtSetLastRetentionTs(STableMgt *pMgt, int64_t retention) {
  if (pMgt == NULL) return 0;

  pMgt->retionTs = retention;
  return 0;
}

int32_t bseTableMgtCreateCache(STableMgt *pMgt) {
  int32_t code = 0;
  int32_t lino = 0;

  SCacheMgt *pCacheMgt = taosMemCalloc(1, sizeof(SCacheMgt));
  if (pCacheMgt == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _error);
  }
  taosThreadRwlockInit(&pCacheMgt->mutex, NULL);

  code = blockCacheOpen(48, blockFree, &pCacheMgt->pBlockCache);

_error:
  return code;
}

int32_t createSubTableMgt(int64_t retenTs, int32_t readOnly, STableMgt *pMgt, SSubTableMgt **pSubMgt) {
  int32_t code = 0;
  int32_t lino = 0;

  SSubTableMgt *p = taosMemCalloc(1, sizeof(SSubTableMgt));
  if (p == NULL) {
    code = terrno;
    TSDB_CHECK_CODE(terrno, lino, _error);
  }

  if (!readOnly) {
    code = tableBuilderMgtInit(p->pBuilderMgt, pMgt->pBse, retenTs);
    TSDB_CHECK_CODE(code, lino, _error);

    p->pBuilderMgt->pMgt = p;
  }

  code = tableReaderMgtInit(p->pReaderMgt, pMgt->pBse, retenTs);
  TSDB_CHECK_CODE(code, lino, _error);

  p->pReaderMgt->pMgt = p;

  code = tableMetaMgtInit(p->pTableMetaMgt, pMgt->pBse, retenTs);
  TSDB_CHECK_CODE(code, lino, _error);

  p->pTableMetaMgt->pMgt = p;

  *pSubMgt = p;
_error:
  if (code != 0) {
    bseError("failed to create sub table mgt at line %d since %s", lino, tstrerror(code));
    destroySubTableMgt(p);
  }
  return code;
}
void destroySubTableMgt(SSubTableMgt *p) {
  if (p != NULL) {
    tableBuilderMgtDestroy(p->pBuilderMgt);
    tableReaderMgtDestroy(p->pReaderMgt);
    tableMetaMgtDestroy(p->pTableMetaMgt);
  }
  taosMemoryFree(p);
}
int32_t bseTableMgtGet(STableMgt *pMgt, int64_t seq, uint8_t **pValue, int32_t *len) {
  if (pMgt == NULL) return 0;

  int32_t code = 0;
  int32_t lino = 0;
  int32_t       readOnly = 1;
  SSubTableMgt *pSubMgt = NULL;

  SBse *pBse = pMgt->pBse;

  int64_t retenTs = 0;
  code = bseGetRetentionTsBySeq(pMgt->pBse, seq, &retenTs);
  TSDB_CHECK_CODE(code, lino, _error);

  if (retenTs > 0) {
    SSubTableMgt **ppSubMgt = taosHashGet(pMgt->pHashObj, &retenTs, sizeof(retenTs));
    if (ppSubMgt == NULL || *ppSubMgt == NULL) {
      code = createSubTableMgt(retenTs, 0, pMgt, &pSubMgt);
      TSDB_CHECK_CODE(code, lino, _error);

      code = taosHashPut(pMgt->pHashObj, &retenTs, sizeof(retenTs), &pSubMgt, sizeof(SSubTableMgt *));
      TSDB_CHECK_CODE(code, lino, _error);

    } else {
      pSubMgt = *ppSubMgt;
    }
  } else {
    pSubMgt = pMgt->pCurrTableMgt;
    if (pSubMgt == NULL) {
      return code;
    }
    readOnly = 0;
  }

  if (readOnly) {
    code = tableReaderMgtSeek(pSubMgt->pReaderMgt, seq, pValue, len);
    TSDB_CHECK_CODE(code, lino, _error);
  } else {
    code = tableBuilderMgtSeek(pSubMgt->pBuilderMgt, seq, pValue, len);
    if (code == TSDB_CODE_OUT_OF_RANGE) {
      code = tableReaderMgtSeek(pSubMgt->pReaderMgt, seq, pValue, len);
      TSDB_CHECK_CODE(code, lino, _error);
    }
  }
_error:
  if (code != 0) {
    bseError("failed to get table at line %d since %s", lino, tstrerror(code));
  }
  return code;
}

int32_t bseTableMgtRecoverTable(STableMgt *pMgt, SBseLiveFileInfo *pInfo) {
  int32_t code = 0;
  int32_t lino = 0;
  if (pMgt == NULL) return 0;

  SSubTableMgt *pSubMgt = NULL;

  code = createSubTableMgt(pInfo->retentionTs, 0, pMgt, &pSubMgt);
  TSDB_CHECK_CODE(code, lino, _error);

  code = tableBuilderMgtRecoverTable(pSubMgt->pBuilderMgt, 0, NULL, pInfo->size);
  TSDB_CHECK_CODE(code, lino, _error);

_error:
  if (code != 0) {
    bseError("failed to recover table at line %d since %s", lino, tstrerror(code));
  }
  destroySubTableMgt(pSubMgt);
  return 0;
}

int32_t bseTableMgtCleanup(void *pMgt) {
  if (pMgt == NULL) return 0;

  STableMgt *p = (STableMgt *)pMgt;

  void *pIter = taosHashIterate(p->pHashObj, NULL);
  while (pIter) {
    SSubTableMgt **ppSubMgt = pIter;
    destroySubTableMgt(*ppSubMgt);
    pIter = taosHashIterate(p->pHashObj, pIter);
  }

  destroySubTableMgt(p->pCurrTableMgt);

  taosHashCleanup(p->pHashObj);
  taosMemoryFree(p);
  return 0;
}

int32_t bseTableMgtAppend(STableMgt *pMgt, SBseBatch *pBatch) {
  int32_t code = 0;
  int32_t lino = 0;
  int32_t retionTs = 0;

  SBse         *pBse = pMgt->pBse;
  SSubTableMgt *pSubMgt = pMgt->pCurrTableMgt;

  if (pSubMgt == NULL) {
    if (pMgt->retionTs != 0) {
      retionTs = pMgt->retionTs;
    } else {
      retionTs = taosGetTimestampSec();
    }

    code = createSubTableMgt(retionTs, 0, pMgt, &pMgt->pCurrTableMgt);
    TSDB_CHECK_CODE(code, lino, _error);
    pSubMgt = pMgt->pCurrTableMgt;
  }

  code = tableBuilderMgtPutBatch(pSubMgt->pBuilderMgt, pBatch);
  TSDB_CHECK_CODE(code, lino, _error);

_error:
  if (code != 0) {
    bseError("failed to append table at line %d since %s", lino, tstrerror(code));
  }
  return code;
}

int32_t bseTableMgtGetLiveFileSet(STableMgt *pMgt, SArray **pList) {
  int32_t code = 0;
  return code;
}

int32_t bseTableMgtCommit(STableMgt *pMgt, SBseLiveFileInfo *pInfo) {
  int32_t code = 0;
  int32_t lino = 0;
  int8_t  flushIdx = -1;

  SSubTableMgt *pSubMgt = pMgt->pCurrTableMgt;
  if (pSubMgt == NULL) {
    bseInfo("nothing to commit table");
    return code;
  }

  code = tableBuilderMgtCommit(pSubMgt->pBuilderMgt, pInfo);
  TSDB_CHECK_CODE(code, lino, _error);
_error:
  if (code != 0) {
    bseError("failed to commit table at line %d since %s", lino, tstrerror(code));
  } else {
    bseInfo("succ to commit table");
  }
  return code;
}

int32_t bseTableMgtUpdateLiveFileSet(STableMgt *pMgt, SArray *pLiveFileSet) {
  int32_t code = 0;
  return code;
}

int32_t bseTableMgtSetBlockCacheSize(STableMgt *pMgt, int32_t cap) {
  int32_t code = 0;
  return code;
  // return blockCacheResize(pMgt->pReaderMgt->pBlockCache, cap);
}

int32_t bseTableMgtSetTableCacheSize(STableMgt *pMgt, int32_t cap) {
  int32_t code = 0;
  return code;
  // return tableCacheResize(pMgt->pReaderMgt->pTableCache, cap);
}

int32_t bseTableMgtClear(STableMgt *pMgt) {
  int32_t code = 0;
  int32_t lino = 0;
  if (pMgt == NULL) return 0;

  destroySubTableMgt(pMgt->pCurrTableMgt);

  void *pIter = taosHashIterate(pMgt->pHashObj, NULL);
  while (pIter) {
    SSubTableMgt **ppSubMgt = pIter;
    destroySubTableMgt(*ppSubMgt);
    pIter = taosHashIterate(pMgt->pHashObj, pIter);
  }
  taosHashClear(pMgt->pHashObj);

_error:
  if (code != 0) {
    bseError("failed to clear table at line %d since %s", lino, tstrerror(code));
  }
  return code;
}

void tableReaderFree(void *pReader) {
  STableReader *p = (STableReader *)pReader;
  if (p != NULL) {
    tableReaderClose(p);
  }
}
void blockFree(void *pBlock) { taosMemoryFree(pBlock); }

int32_t tableReaderMgtInit(STableReaderMgt *pReader, SBse *pBse, int64_t retention) {
  int32_t code = 0;
  int32_t lino = 0;

  taosThreadRwlockInit(&pReader->mutex, NULL);

  code = blockCacheOpen(48, blockFree, &pReader->pBlockCache);
  TSDB_CHECK_CODE(code, lino, _error);

  code = tableCacheOpen(32, tableReaderFree, &pReader->pTableCache);
  TSDB_CHECK_CODE(code, lino, _error);

  pReader->pBse = pBse;
  pReader->retenTs = retention;

_error:
  if (code != 0) {
    bseError("failed to init table pReaderMgt mgt at line %d since %s", lino, tstrerror(code));
  }
  return code;
}
void tableReaderMgtSetRetion(STableReaderMgt *pReader, int64_t retention) { pReader->retenTs = retention; }

int32_t tableReaderMgtClear(STableReaderMgt *pReader) {
  int32_t code = 0;

  taosThreadRwlockWrlock(&pReader->mutex);

  (void)(tableCacheClear(pReader->pTableCache));

  (void)(blockCacheClear(pReader->pBlockCache));
  taosThreadRwlockUnlock(&pReader->mutex);

  return code;
}

void tableReaderMgtDestroy(STableReaderMgt *pReader) {
  tableCacheClose(pReader->pTableCache);
  blockCacheClose(pReader->pBlockCache);
  taosThreadRwlockDestroy(&pReader->mutex);
}

int32_t tableReaderMgtSeek(STableReaderMgt *pReaderMgt, int64_t seq, uint8_t **pValue, int32_t *len) {
  int32_t code = 0;
  int32_t lino = 0;

  STableReader    *pReader = NULL;

  code = tableReaderOpen(pReaderMgt->retenTs, &pReader, pReaderMgt);
  TSDB_CHECK_CODE(code, lino, _error);

  code = tableReaderGet(pReader, seq, pValue, len);
  TSDB_CHECK_CODE(code, lino, _error);

_error:
  if (code != 0) {
    bseError("failed to seek table pReaderMgt at line %d since %s", lino, tstrerror(code));
  }

  tableReaderClose(pReader);
  return code;
}

int32_t tableBuilderMgtInit(STableBuilderMgt *pMgt, SBse *pBse, int64_t retention) {
  int32_t code = 0;
  int32_t lino = 0;

  taosThreadMutexInit(&pMgt->mutex, NULL);
  pMgt->pBse = pBse;

  for (int32_t i = 0; i < 2; i++) {
    pMgt->p[i] = NULL;
  }
  pMgt->inUse = 0;
  pMgt->retenTs = retention;
  return code;
}

int32_t tableBuilderMgtClear(STableBuilderMgt *pMgt) {
  int32_t code = 0;
  int32_t lino = 0;

  taosThreadMutexLock(&pMgt->mutex);
  for (int32_t i = 0; i < 2; i++) {
    if (pMgt->p[i] != NULL) {
      tableBuilderClose(pMgt->p[i], 0);
      pMgt->p[i] = NULL;
      pMgt->inUse = 0;
    }
  }
  taosThreadMutexUnlock(&pMgt->mutex);
  return code;
}

int32_t tableBuilderMgtPutBatch(STableBuilderMgt *pMgt, SBseBatch *pBatch) {
  int32_t code = 0;
  int32_t lino = 0;
  int64_t seq = pBatch->startSeq;
  taosThreadMutexLock(&pMgt->mutex);
  STableBuilder *p = pMgt->p[pMgt->inUse];
  taosThreadMutexUnlock(&pMgt->mutex);

  if (p == NULL) {
    code = tableBuilderMgtGetBuilder(pMgt, seq, &p);
    TSDB_CHECK_CODE(code, lino, _error);
  }

  code = tableBuilderPutBatch(p, pBatch);
  TSDB_CHECK_CODE(code, lino, _error);
_error:
  if (code != 0) {
    bseError("failed to put batch to table builder at line %d since %s", lino, tstrerror(code));
  }
  return code;
}

int32_t tableBuilderMgtSeek(STableBuilderMgt *pMgt, int64_t seq, uint8_t **pValue, int32_t *len) {
  int32_t        code = 0;
  int32_t        lino = 0;
  STableBuilder *pBuilder = NULL;

  taosThreadMutexLock(&pMgt->mutex);
  int8_t inUse = pMgt->inUse;
  pBuilder = pMgt->p[inUse];
  taosThreadMutexUnlock(&pMgt->mutex);

  if (pBuilder && seqRangeContains(&pBuilder->tableRange, seq)) {
    code = tableBuilderGet(pBuilder, seq, pValue, len);
  } else {
    code = TSDB_CODE_OUT_OF_RANGE;  //  continue to read from reader
  }
  return code;
}

int32_t tableBuilderMgtGetBuilder(STableBuilderMgt *pMgt, int64_t seq, STableBuilder **pBuilder) {
  int32_t code = 0;
  int32_t lino = 0;

  SBse *pBse = pMgt->pBse;

  STableBuilder *p = NULL;

  code = tableBuilderOpen(pMgt->retenTs, &p, pBse);
  TSDB_CHECK_CODE(code, lino, _error);

  p->pTableMeta = pMgt->pMgt->pTableMetaMgt->pTableMeta;

  p->pBse = pMgt->pBse;
  pMgt->p[pMgt->inUse] = p;

  *pBuilder = p;

_error:
  if (code != 0) {
    bseError("failed to open table builder at line %d since %s", __LINE__, tstrerror(code));
  }

  return code;
}

int32_t tableBuilderMgtRecoverTable(STableBuilderMgt *pMgt, int64_t seq, STableBuilder **pBuilder, int64_t size) {
  int32_t        code = 0;
  int32_t        lino = 0;
  STableBuilder *pTable = NULL;

  code = tableBuilderMgtGetBuilder(pMgt, seq, &pTable);
  TSDB_CHECK_CODE(code, lino, _error);

  if (pTable->offset > size) {
    code = tableBuilderTruncFile(pTable, size);
    TSDB_CHECK_CODE(code, lino, _error);
  }
_error:
  if (code != 0) {
    bseError("failed to open table builder at line %d since %s", lino, tstrerror(code));
  }

  return code;
}
int32_t tableBuilderMgtCommit(STableBuilderMgt *pMgt, SBseLiveFileInfo *pInfo) {
  int32_t        code = 0;
  int32_t        lino = 0;
  int8_t         flushIdx = -1;
  STableBuilder *pBuilder = NULL;

  taosThreadMutexLock(&pMgt->mutex);
  pBuilder = pMgt->p[pMgt->inUse];

  taosThreadMutexUnlock(&pMgt->mutex);
  if (pBuilder != NULL) {
    code = tableBuilderCommit(pBuilder, pInfo);
    TSDB_CHECK_CODE(code, lino, _error);
  }
_error:
  if (code != 0) {
    bseError("failed to commit table builder at line %d since %s", lino, tstrerror(code));
  }
  return code;
}

void tableBuilderMgtDestroy(STableBuilderMgt *pMgt) {
  for (int32_t i = 0; i < 2; i++) {
    if (pMgt->p[i] != NULL) {
      tableBuilderClose(pMgt->p[i], 0);
    }
  }
  taosThreadMutexDestroy(&pMgt->mutex);
}

int32_t tableMetaMgtInit(STableMetaMgt *pMgt, SBse *pBse, int64_t retention) {
  int32_t code = 0;
  int32_t lino = 0;
  pMgt->pBse = pBse;

  code = tableMetaOpen(NULL, &pMgt->pTableMeta, pMgt);
  TSDB_CHECK_CODE(code, lino, _error);

  pMgt->retenTs = retention;
  pMgt->pTableMeta->retentionTs = retention;
  pMgt->pTableMeta->pBse = pBse;

_error:
  if (code != 0) {
    bseError("failed to init table meta mgt at line %d since %s", lino, tstrerror(code));
  }
  return code;
}

static void tableMetaMgtDestroy(STableMetaMgt *pMgt) {
  if (pMgt->pTableMeta != NULL) {
    tableMetaClose(pMgt->pTableMeta);
    pMgt->pTableMeta = NULL;
  }
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "bseTableMgt.h"
17	#include "bse.h"
18	#include "bseTable.h"
19	#include "bseUtil.h"
20	#include "thash.h"
21
22	static int32_t tableReaderMgtInit(STableReaderMgt pReader, SBse pBse, int64_t retention);
23	static void tableReaderMgtSetRetion(STableReaderMgt *pReader, int64_t retention);
24	static int32_t tableReaderMgtSeek(STableReaderMgt pReaderMgt, int64_t seq, uint8_t pValue, int32_t len);
25	static int32_t tableReaderMgtClear(STableReaderMgt *pReader);
26	static void tableReaderMgtDestroy(STableReaderMgt *pReader);
27
28	static int32_t tableBuilderMgtInit(STableBuilderMgt pMgt, SBse pBse, int64_t retention);
29	static void tableBuilderMgtSetRetion(STableBuilderMgt *pMgt, int64_t retention);
30	static int32_t tableBuilderMgtGetBuilder(STableBuilderMgt pMgt, int64_t seq, STableBuilder *p);
31	static int32_t tableBuilderMgtCommit(STableBuilderMgt pMgt, SBseLiveFileInfo pInfo);
32	static int32_t tableBuilderMgtSeek(STableBuilderMgt pMgt, int64_t seq, uint8_t pValue, int32_t len);
33	static int32_t tableBuilderMgtPutBatch(STableBuilderMgt pMgt, SBseBatch pBatch);
34	static int32_t tableBuilderMgtClear(STableBuilderMgt *pMgt);
35	static void tableBuilderMgtDestroy(STableBuilderMgt *pMgt);
36
37	static int32_t tableBuilderMgtRecoverTable(STableBuilderMgt pMgt, int64_t seq, STableBuilder *pBuilder, int64_t size);
38
39	static int32_t tableMetaMgtInit(STableMetaMgt pMgt, SBse pBse, int64_t retention);
40	static void tableMetaMgtDestroy(STableMetaMgt *pMgt);
41
42	static void tableReaderFree(void *pReader);
43
44	static void blockFree(void *pBlock);
45
46	int32_t bseTableMgtCreate(SBse pBse, void *pMgt) {	9,748✔
47	int32_t code = 0;	9,748✔
48	int32_t lino = 0;	9,748✔
49
50	STableMgt *p = taosMemoryCalloc(1, sizeof(STableMgt));	9,748!
51	if (p == NULL) {	9,776!
52	return terrno;	×
53	}
54	p->pBse = pBse;	9,776✔
55	p->pHashObj = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);	9,776✔
56	if (p->pHashObj == NULL) {	9,776!
57	TSDB_CHECK_CODE(code = terrno, lino, _error);	×
58	}
59
60	*pMgt = p;	9,776✔
61	_error:	9,776✔
62	if (code != 0) {	9,776!
63	if (p != NULL)	×
64	bseError("vgId:%d failed to open table pBuilderMgt at line %d since %s", BSE_GET_VGID((SBse *)p->pBse), lino,	×
65	tstrerror(code));
66	bseTableMgtCleanup(p);	×
67	}
68	return code;	9,776✔
69	}
70
71	int32_t bseTableMgtSetLastRetentionTs(STableMgt *pMgt, int64_t retention) {	×
72	if (pMgt == NULL) return 0;	×
73
74	pMgt->retionTs = retention;	×
75	return 0;	×
76	}
77
78	int32_t bseTableMgtCreateCache(STableMgt *pMgt) {	×
79	int32_t code = 0;	×
80	int32_t lino = 0;	×
81
82	SCacheMgt *pCacheMgt = taosMemCalloc(1, sizeof(SCacheMgt));	×
83	if (pCacheMgt == NULL) {	×
84	TSDB_CHECK_CODE(code = terrno, lino, _error);	×
85	}
86	taosThreadRwlockInit(&pCacheMgt->mutex, NULL);	×
87
88	code = blockCacheOpen(48, blockFree, &pCacheMgt->pBlockCache);	×
89
90	_error:	×
91	return code;	×
92	}
93
94	int32_t createSubTableMgt(int64_t retenTs, int32_t readOnly, STableMgt pMgt, SSubTableMgt *pSubMgt) {	×
95	int32_t code = 0;	×
96	int32_t lino = 0;	×
97
98	SSubTableMgt *p = taosMemCalloc(1, sizeof(SSubTableMgt));	×
99	if (p == NULL) {	×
100	code = terrno;	×
101	TSDB_CHECK_CODE(terrno, lino, _error);	×
102	}
103
104	if (!readOnly) {	×
105	code = tableBuilderMgtInit(p->pBuilderMgt, pMgt->pBse, retenTs);	×
106	TSDB_CHECK_CODE(code, lino, _error);	×
107
108	p->pBuilderMgt->pMgt = p;	×
109	}
110
111	code = tableReaderMgtInit(p->pReaderMgt, pMgt->pBse, retenTs);	×
112	TSDB_CHECK_CODE(code, lino, _error);	×
113
114	p->pReaderMgt->pMgt = p;	×
115
116	code = tableMetaMgtInit(p->pTableMetaMgt, pMgt->pBse, retenTs);	×
117	TSDB_CHECK_CODE(code, lino, _error);	×
118
119	p->pTableMetaMgt->pMgt = p;	×
120
121	*pSubMgt = p;	×
122	_error:	×
123	if (code != 0) {	×
124	bseError("failed to create sub table mgt at line %d since %s", lino, tstrerror(code));	×
125	destroySubTableMgt(p);	×
126	}
127	return code;	×
128	}
129	void destroySubTableMgt(SSubTableMgt *p) {	9,777✔
130	if (p != NULL) {	9,777!
131	tableBuilderMgtDestroy(p->pBuilderMgt);	×
132	tableReaderMgtDestroy(p->pReaderMgt);	×
133	tableMetaMgtDestroy(p->pTableMetaMgt);	×
134	}
135	taosMemoryFree(p);	9,777!
136	}	9,777✔
137	int32_t bseTableMgtGet(STableMgt pMgt, int64_t seq, uint8_t pValue, int32_t len) {	×
138	if (pMgt == NULL) return 0;	×
139
140	int32_t code = 0;	×
141	int32_t lino = 0;	×
142	int32_t readOnly = 1;	×
143	SSubTableMgt *pSubMgt = NULL;	×
144
145	SBse *pBse = pMgt->pBse;	×
146
147	int64_t retenTs = 0;	×
148	code = bseGetRetentionTsBySeq(pMgt->pBse, seq, &retenTs);	×
149	TSDB_CHECK_CODE(code, lino, _error);	×
150
151	if (retenTs > 0) {	×
152	SSubTableMgt **ppSubMgt = taosHashGet(pMgt->pHashObj, &retenTs, sizeof(retenTs));	×
153	if (ppSubMgt == NULL \|\| *ppSubMgt == NULL) {	×
154	code = createSubTableMgt(retenTs, 0, pMgt, &pSubMgt);	×
155	TSDB_CHECK_CODE(code, lino, _error);	×
156
157	code = taosHashPut(pMgt->pHashObj, &retenTs, sizeof(retenTs), &pSubMgt, sizeof(SSubTableMgt *));	×
158	TSDB_CHECK_CODE(code, lino, _error);	×
159
160	} else {
161	pSubMgt = *ppSubMgt;	×
162	}
163	} else {
164	pSubMgt = pMgt->pCurrTableMgt;	×
165	if (pSubMgt == NULL) {	×
166	return code;	×
167	}
168	readOnly = 0;	×
169	}
170
171	if (readOnly) {	×
172	code = tableReaderMgtSeek(pSubMgt->pReaderMgt, seq, pValue, len);	×
173	TSDB_CHECK_CODE(code, lino, _error);	×
174	} else {
175	code = tableBuilderMgtSeek(pSubMgt->pBuilderMgt, seq, pValue, len);	×
176	if (code == TSDB_CODE_OUT_OF_RANGE) {	×
177	code = tableReaderMgtSeek(pSubMgt->pReaderMgt, seq, pValue, len);	×
178	TSDB_CHECK_CODE(code, lino, _error);	×
179	}
180	}
181	_error:	×
182	if (code != 0) {	×
183	bseError("failed to get table at line %d since %s", lino, tstrerror(code));	×
184	}
185	return code;	×
186	}
187
188	int32_t bseTableMgtRecoverTable(STableMgt pMgt, SBseLiveFileInfo pInfo) {	×
189	int32_t code = 0;	×
190	int32_t lino = 0;	×
191	if (pMgt == NULL) return 0;	×
192
193	SSubTableMgt *pSubMgt = NULL;	×
194
195	code = createSubTableMgt(pInfo->retentionTs, 0, pMgt, &pSubMgt);	×
196	TSDB_CHECK_CODE(code, lino, _error);	×
197
198	code = tableBuilderMgtRecoverTable(pSubMgt->pBuilderMgt, 0, NULL, pInfo->size);	×
199	TSDB_CHECK_CODE(code, lino, _error);	×
200
201	_error:	×
202	if (code != 0) {	×
203	bseError("failed to recover table at line %d since %s", lino, tstrerror(code));	×
204	}
205	destroySubTableMgt(pSubMgt);	×
206	return 0;	×
207	}
208
209	int32_t bseTableMgtCleanup(void *pMgt) {	9,777✔
210	if (pMgt == NULL) return 0;	9,777!
211
212	STableMgt p = (STableMgt )pMgt;	9,777✔
213
214	void *pIter = taosHashIterate(p->pHashObj, NULL);	9,777✔
215	while (pIter) {	9,777!
216	SSubTableMgt **ppSubMgt = pIter;	×
217	destroySubTableMgt(*ppSubMgt);	×
218	pIter = taosHashIterate(p->pHashObj, pIter);	×
219	}
220
221	destroySubTableMgt(p->pCurrTableMgt);	9,777✔
222
223	taosHashCleanup(p->pHashObj);	9,777✔
224	taosMemoryFree(p);	9,776!
225	return 0;	9,776✔
226	}
227
228	int32_t bseTableMgtAppend(STableMgt pMgt, SBseBatch pBatch) {	×
229	int32_t code = 0;	×
230	int32_t lino = 0;	×
231	int32_t retionTs = 0;	×
232
233	SBse *pBse = pMgt->pBse;	×
234	SSubTableMgt *pSubMgt = pMgt->pCurrTableMgt;	×
235
236	if (pSubMgt == NULL) {	×
237	if (pMgt->retionTs != 0) {	×
238	retionTs = pMgt->retionTs;	×
239	} else {
240	retionTs = taosGetTimestampSec();	×
241	}
242
243	code = createSubTableMgt(retionTs, 0, pMgt, &pMgt->pCurrTableMgt);	×
244	TSDB_CHECK_CODE(code, lino, _error);	×
245	pSubMgt = pMgt->pCurrTableMgt;	×
246	}
247
248	code = tableBuilderMgtPutBatch(pSubMgt->pBuilderMgt, pBatch);	×
249	TSDB_CHECK_CODE(code, lino, _error);	×
250
251	_error:	×
252	if (code != 0) {	×
253	bseError("failed to append table at line %d since %s", lino, tstrerror(code));	×
254	}
255	return code;	×
256	}
257
258	int32_t bseTableMgtGetLiveFileSet(STableMgt pMgt, SArray *pList) {	×
259	int32_t code = 0;	×
260	return code;	×
261	}
262
263	int32_t bseTableMgtCommit(STableMgt pMgt, SBseLiveFileInfo pInfo) {	11,354✔
264	int32_t code = 0;	11,354✔
265	int32_t lino = 0;	11,354✔
266	int8_t flushIdx = -1;	11,354✔
267
268	SSubTableMgt *pSubMgt = pMgt->pCurrTableMgt;	11,354✔
269	if (pSubMgt == NULL) {	11,354!
270	bseInfo("nothing to commit table");	11,354!
271	return code;	11,354✔
272	}
273
274	code = tableBuilderMgtCommit(pSubMgt->pBuilderMgt, pInfo);	×
275	TSDB_CHECK_CODE(code, lino, _error);	×
276	_error:	×
277	if (code != 0) {	×
278	bseError("failed to commit table at line %d since %s", lino, tstrerror(code));	×
279	} else {
280	bseInfo("succ to commit table");	×
281	}
282	return code;	×
283	}
284
285	int32_t bseTableMgtUpdateLiveFileSet(STableMgt pMgt, SArray pLiveFileSet) {	×
286	int32_t code = 0;	×
287	return code;	×
288	}
289
290	int32_t bseTableMgtSetBlockCacheSize(STableMgt *pMgt, int32_t cap) {	×
291	int32_t code = 0;	×
292	return code;	×
293	// return blockCacheResize(pMgt->pReaderMgt->pBlockCache, cap);
294	}
295
296	int32_t bseTableMgtSetTableCacheSize(STableMgt *pMgt, int32_t cap) {	×
297	int32_t code = 0;	×
298	return code;	×
299	// return tableCacheResize(pMgt->pReaderMgt->pTableCache, cap);
300	}
301
302	int32_t bseTableMgtClear(STableMgt *pMgt) {	×
303	int32_t code = 0;	×
304	int32_t lino = 0;	×
305	if (pMgt == NULL) return 0;	×
306
307	destroySubTableMgt(pMgt->pCurrTableMgt);	×
308
309	void *pIter = taosHashIterate(pMgt->pHashObj, NULL);	×
310	while (pIter) {	×
311	SSubTableMgt **ppSubMgt = pIter;	×
312	destroySubTableMgt(*ppSubMgt);	×
313	pIter = taosHashIterate(pMgt->pHashObj, pIter);	×
314	}
315	taosHashClear(pMgt->pHashObj);	×
316
317	_error:	×
318	if (code != 0) {	×
319	bseError("failed to clear table at line %d since %s", lino, tstrerror(code));	×
320	}
321	return code;	×
322	}
323
324	void tableReaderFree(void *pReader) {	×
325	STableReader p = (STableReader )pReader;	×
326	if (p != NULL) {	×
327	tableReaderClose(p);	×
328	}
329	}	×
330	void blockFree(void *pBlock) { taosMemoryFree(pBlock); }	×
331
332	int32_t tableReaderMgtInit(STableReaderMgt pReader, SBse pBse, int64_t retention) {	×
333	int32_t code = 0;	×
334	int32_t lino = 0;	×
335
336	taosThreadRwlockInit(&pReader->mutex, NULL);	×
337
338	code = blockCacheOpen(48, blockFree, &pReader->pBlockCache);	×
339	TSDB_CHECK_CODE(code, lino, _error);	×
340
341	code = tableCacheOpen(32, tableReaderFree, &pReader->pTableCache);	×
342	TSDB_CHECK_CODE(code, lino, _error);	×
343
344	pReader->pBse = pBse;	×
345	pReader->retenTs = retention;	×
346
347	_error:	×
348	if (code != 0) {	×
349	bseError("failed to init table pReaderMgt mgt at line %d since %s", lino, tstrerror(code));	×
350	}
351	return code;	×
352	}
353	void tableReaderMgtSetRetion(STableReaderMgt *pReader, int64_t retention) { pReader->retenTs = retention; }	×
354
355	int32_t tableReaderMgtClear(STableReaderMgt *pReader) {	×
356	int32_t code = 0;	×
357
358	taosThreadRwlockWrlock(&pReader->mutex);	×
359
360	(void)(tableCacheClear(pReader->pTableCache));	×
361
362	(void)(blockCacheClear(pReader->pBlockCache));	×
363	taosThreadRwlockUnlock(&pReader->mutex);	×
364
365	return code;	×
366	}
367
368	void tableReaderMgtDestroy(STableReaderMgt *pReader) {	×
369	tableCacheClose(pReader->pTableCache);	×
370	blockCacheClose(pReader->pBlockCache);	×
371	taosThreadRwlockDestroy(&pReader->mutex);	×
372	}	×
373
374	int32_t tableReaderMgtSeek(STableReaderMgt pReaderMgt, int64_t seq, uint8_t pValue, int32_t len) {	×
375	int32_t code = 0;	×
376	int32_t lino = 0;	×
377
378	STableReader *pReader = NULL;	×
379
380	code = tableReaderOpen(pReaderMgt->retenTs, &pReader, pReaderMgt);	×
381	TSDB_CHECK_CODE(code, lino, _error);	×
382
383	code = tableReaderGet(pReader, seq, pValue, len);	×
384	TSDB_CHECK_CODE(code, lino, _error);	×
385
386	_error:	×
387	if (code != 0) {	×
388	bseError("failed to seek table pReaderMgt at line %d since %s", lino, tstrerror(code));	×
389	}
390
391	tableReaderClose(pReader);	×
392	return code;	×
393	}
394
395	int32_t tableBuilderMgtInit(STableBuilderMgt pMgt, SBse pBse, int64_t retention) {	×
396	int32_t code = 0;	×
397	int32_t lino = 0;	×
398
399	taosThreadMutexInit(&pMgt->mutex, NULL);	×
400	pMgt->pBse = pBse;	×
401
402	for (int32_t i = 0; i < 2; i++) {	×
403	pMgt->p[i] = NULL;	×
404	}
405	pMgt->inUse = 0;	×
406	pMgt->retenTs = retention;	×
407	return code;	×
408	}
409
410	int32_t tableBuilderMgtClear(STableBuilderMgt *pMgt) {	×
411	int32_t code = 0;	×
412	int32_t lino = 0;	×
413
414	taosThreadMutexLock(&pMgt->mutex);	×
415	for (int32_t i = 0; i < 2; i++) {	×
416	if (pMgt->p[i] != NULL) {	×
417	tableBuilderClose(pMgt->p[i], 0);	×
418	pMgt->p[i] = NULL;	×
419	pMgt->inUse = 0;	×
420	}
421	}
422	taosThreadMutexUnlock(&pMgt->mutex);	×
423	return code;	×
424	}
425
426	int32_t tableBuilderMgtPutBatch(STableBuilderMgt pMgt, SBseBatch pBatch) {	×
427	int32_t code = 0;	×
428	int32_t lino = 0;	×
429	int64_t seq = pBatch->startSeq;	×
430	taosThreadMutexLock(&pMgt->mutex);	×
431	STableBuilder *p = pMgt->p[pMgt->inUse];	×
432	taosThreadMutexUnlock(&pMgt->mutex);	×
433
434	if (p == NULL) {	×
435	code = tableBuilderMgtGetBuilder(pMgt, seq, &p);	×
436	TSDB_CHECK_CODE(code, lino, _error);	×
437	}
438
439	code = tableBuilderPutBatch(p, pBatch);	×
440	TSDB_CHECK_CODE(code, lino, _error);	×
441	_error:	×
442	if (code != 0) {	×
443	bseError("failed to put batch to table builder at line %d since %s", lino, tstrerror(code));	×
444	}
445	return code;	×
446	}
447
448	int32_t tableBuilderMgtSeek(STableBuilderMgt pMgt, int64_t seq, uint8_t pValue, int32_t len) {	×
449	int32_t code = 0;	×
450	int32_t lino = 0;	×
451	STableBuilder *pBuilder = NULL;	×
452
453	taosThreadMutexLock(&pMgt->mutex);	×
454	int8_t inUse = pMgt->inUse;	×
455	pBuilder = pMgt->p[inUse];	×
456	taosThreadMutexUnlock(&pMgt->mutex);	×
457
458	if (pBuilder && seqRangeContains(&pBuilder->tableRange, seq)) {	×
459	code = tableBuilderGet(pBuilder, seq, pValue, len);	×
460	} else {
461	code = TSDB_CODE_OUT_OF_RANGE; // continue to read from reader	×
462	}
463	return code;	×
464	}
465
466	int32_t tableBuilderMgtGetBuilder(STableBuilderMgt pMgt, int64_t seq, STableBuilder *pBuilder) {	×
467	int32_t code = 0;	×
468	int32_t lino = 0;	×
469
470	SBse *pBse = pMgt->pBse;	×
471
472	STableBuilder *p = NULL;	×
473
474	code = tableBuilderOpen(pMgt->retenTs, &p, pBse);	×
475	TSDB_CHECK_CODE(code, lino, _error);	×
476
477	p->pTableMeta = pMgt->pMgt->pTableMetaMgt->pTableMeta;	×
478
479	p->pBse = pMgt->pBse;	×
480	pMgt->p[pMgt->inUse] = p;	×
481
482	*pBuilder = p;	×
483
484	_error:	×
485	if (code != 0) {	×
486	bseError("failed to open table builder at line %d since %s", __LINE__, tstrerror(code));	×
487	}
488
489	return code;	×
490	}
491
492	int32_t tableBuilderMgtRecoverTable(STableBuilderMgt pMgt, int64_t seq, STableBuilder *pBuilder, int64_t size) {	×
493	int32_t code = 0;	×
494	int32_t lino = 0;	×
495	STableBuilder *pTable = NULL;	×
496
497	code = tableBuilderMgtGetBuilder(pMgt, seq, &pTable);	×
498	TSDB_CHECK_CODE(code, lino, _error);	×
499
500	if (pTable->offset > size) {	×
501	code = tableBuilderTruncFile(pTable, size);	×
502	TSDB_CHECK_CODE(code, lino, _error);	×
503	}
504	_error:	×
505	if (code != 0) {	×
506	bseError("failed to open table builder at line %d since %s", lino, tstrerror(code));	×
507	}
508
509	return code;	×
510	}
511	int32_t tableBuilderMgtCommit(STableBuilderMgt pMgt, SBseLiveFileInfo pInfo) {	×
512	int32_t code = 0;	×
513	int32_t lino = 0;	×
514	int8_t flushIdx = -1;	×
515	STableBuilder *pBuilder = NULL;	×
516
517	taosThreadMutexLock(&pMgt->mutex);	×
518	pBuilder = pMgt->p[pMgt->inUse];	×
519
520	taosThreadMutexUnlock(&pMgt->mutex);	×
521	if (pBuilder != NULL) {	×
522	code = tableBuilderCommit(pBuilder, pInfo);	×
523	TSDB_CHECK_CODE(code, lino, _error);	×
524	}
525	_error:	×
526	if (code != 0) {	×
527	bseError("failed to commit table builder at line %d since %s", lino, tstrerror(code));	×
528	}
529	return code;	×
530	}
531
532	void tableBuilderMgtDestroy(STableBuilderMgt *pMgt) {	×
533	for (int32_t i = 0; i < 2; i++) {	×
534	if (pMgt->p[i] != NULL) {	×
535	tableBuilderClose(pMgt->p[i], 0);	×
536	}
537	}
538	taosThreadMutexDestroy(&pMgt->mutex);	×
539	}	×
540
541	int32_t tableMetaMgtInit(STableMetaMgt pMgt, SBse pBse, int64_t retention) {	×
542	int32_t code = 0;	×
543	int32_t lino = 0;	×
544	pMgt->pBse = pBse;	×
545
546	code = tableMetaOpen(NULL, &pMgt->pTableMeta, pMgt);	×
547	TSDB_CHECK_CODE(code, lino, _error);	×
548
549	pMgt->retenTs = retention;	×
550	pMgt->pTableMeta->retentionTs = retention;	×
551	pMgt->pTableMeta->pBse = pBse;	×
552
553	_error:	×
554	if (code != 0) {	×
555	bseError("failed to init table meta mgt at line %d since %s", lino, tstrerror(code));	×
556	}
557	return code;	×
558	}
559
560	static void tableMetaMgtDestroy(STableMetaMgt *pMgt) {	×
561	if (pMgt->pTableMeta != NULL) {	×
562	tableMetaClose(pMgt->pTableMeta);	×
563	pMgt->pTableMeta = NULL;	×
564	}
565	}	×

taosdata / TDengine / #4548

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous