#4987

Committed 16 Mar 2026 12:26PM UTC coverage: 73.883% (+36.6%) from 37.305%

Build # #4987

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat: support secure delete option. (#34591)

Coverage Stats

209 of 391 new or added lines in 24 files covered. (53.45%)

3062 existing lines in 140 files now uncovered.

261133 of 353439 relevant lines covered (73.88%)

121262425.02 hits per line

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

0.0

/source/dnode/vnode/src/bse/bseSnapshot.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 "bseSnapshot.h"
#include "bseInc.h"
#include "bseTable.h"
#include "bseTableMgt.h"
#include "bseUtil.h"
#include "vnodeInt.h"

static int32_t bseRawFileWriterOpen(SBse *pBse, int64_t sver, int64_t ever, SBseSnapMeta *pMeta,
                                    SBseRawFileWriter **pWriter);
static int32_t bseRawFileWriterDoWrite(SBseRawFileWriter *p, uint8_t *data, int32_t len);
static void    bseRawFileWriterClose(SBseRawFileWriter *p, int8_t rollback);
static void    bseRawFileGenLiveInfo(SBseRawFileWriter *p, SBseLiveFileInfo *pInfo);

static int32_t bseSnapReaderReadImpl(SBseSnapReader *p, uint8_t **data, int32_t *len);

static int32_t bseRawFileWriterOpen(SBse *pBse, int64_t sver, int64_t ever, SBseSnapMeta *pMeta,
                                    SBseRawFileWriter **pWriter) {
  int32_t code = 0;
  int32_t lino = 0;

  char name[TSDB_FILENAME_LEN] = {0};
  char path[TSDB_FILENAME_LEN] = {0};

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

  SSeqRange *range = &pMeta->range;
  if (pMeta->fileType == BSE_TABLE_SNAP) {
    bseBuildDataName(pMeta->timestamp, name);
    bseBuildFullName(pBse, name, path);
  } else if (pMeta->fileType == BSE_TABLE_META_TYPE) {
    bseBuildMetaName(pMeta->timestamp, name);
    bseBuildFullName(pBse, name, path);
  } else if (pMeta->fileType == BSE_CURRENT_SNAP) {
    bseBuildCurrentFullName(pBse, path);
  } else {
    return TSDB_CODE_INVALID_MSG;
  }

  p->fileType = pMeta->fileType;
  p->range = *range;
  p->pBse = pBse;
  p->pFile = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND | TD_FILE_TRUNC);
  if (p->pFile == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _error);
  }

  *pWriter = p;
_error:
  if (code) {
    if (p != NULL) {
      bseError("vgId:%d failed to open table pWriter at line %d since %s", BSE_VGID((SBse *)pBse), lino,
               tstrerror(code));
      bseRawFileWriterClose(p, 0);
    }
    *pWriter = NULL;
  }

  return code;
}

int32_t bseRawFileWriterDoWrite(SBseRawFileWriter *p, uint8_t *data, int32_t len) {
  int32_t code = 0;
  int32_t nwrite = taosWriteFile(p->pFile, data, len);
  if (nwrite != len) {
    return terrno;
  }
  p->offset += len;

  return code;
}

void bseRawFileWriterClose(SBseRawFileWriter *p, int8_t rollback) {
  if (p == NULL) return;

  int32_t code = 0;
  if (taosCloseFile(&p->pFile) != 0) {
    bseError("vgId:%d failed to close table pWriter since %s", BSE_VGID((SBse *)p->pBse), tstrerror(terrno));
  }
  taosMemoryFree(p);

  return;
}

void bseRawFileGenLiveInfo(SBseRawFileWriter *p, SBseLiveFileInfo *pInfo) {
  pInfo->range = p->range;
  pInfo->size = p->offset;
  memcpy(pInfo->name, p->name, sizeof(p->name));
}

int32_t bseSnapMayOpenNewFile(SBseSnapWriter *pWriter, SBseSnapMeta *pMeta) {
  int32_t code = 0;

  SBse *pBse = pWriter->pBse;

  SBseRawFileWriter *pOld = pWriter->pWriter;

  if (pOld == NULL || (pOld->fileType != pMeta->fileType)) {
    if (pOld != NULL) {
      SBseLiveFileInfo info;
      bseRawFileGenLiveInfo(pOld, &info);

      if (taosArrayPush(pWriter->pFileSet, &info) == NULL) {
        code = terrno;
        bseError("vgId:%d failed to push file info since %s", BSE_VGID((SBse *)pBse), tstrerror(code));
        return code;
      }
      bseRawFileWriterClose(pOld, 0);
      pWriter->pWriter = NULL;
    }

    SBseRawFileWriter *pNew = NULL;
    code = bseRawFileWriterOpen(pBse, 0, 0, pMeta, &pNew);
    if (code) {
      bseError("vgId:%d failed to open table pWriter since %s", BSE_VGID((SBse *)pBse), tstrerror(code));
      return code;
    }
    pWriter->pWriter = pNew;
  }
  return code;
}

int32_t bseSnapWriterOpen(SBse *pBse, int64_t sver, int64_t ever, SBseSnapWriter **pWriter) {
  int32_t code = 0;
  int32_t lino = 0;

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

  p->pFileSet = taosArrayInit(128, sizeof(SBseLiveFileInfo));
  if (p->pFileSet == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _error);
  }
  p->pBse = pBse;

  *pWriter = p;

_error:
  if (code) {
    if (p != NULL) {
      bseError("vgId:%d failed to open table pWriter at line %d since %s", BSE_VGID((SBse *)pBse), lino,
               tstrerror(code));
      bseSnapWriterClose(&p, 0);
    }
    *pWriter = NULL;
  }

  return code;
}
int32_t bseSnapWriterWrite(SBseSnapWriter *p, uint8_t *data, int32_t len) {
  int32_t       code;
  int32_t       lino = 0;
  SSnapDataHdr *pHdr = (SSnapDataHdr *)data;
  if (pHdr->size + sizeof(SSnapDataHdr) != len) {
    return TSDB_CODE_INVALID_MSG;
  }
  SBseSnapMeta *pMeta = (SBseSnapMeta *)pHdr->data;

  uint8_t *pBuf = pHdr->data + sizeof(SBseSnapMeta);
  int64_t  tlen = len - sizeof(SSnapDataHdr) - sizeof(SBseSnapMeta);

  code = bseSnapMayOpenNewFile(p, pMeta);
  TSDB_CHECK_CODE(code, lino, _error);

  code = bseRawFileWriterDoWrite(p->pWriter, pBuf, tlen);
  TSDB_CHECK_CODE(code, lino, _error);
_error:
  if (code) {
    if (p->pWriter != NULL) {
      bseError("vgId:%d failed to write snapshot data since %s", BSE_VGID((SBse *)p->pBse), tstrerror(code));
      bseRawFileWriterClose(p->pWriter, 0);
    }
    return code;
  }
  return code;
}
void bseSnapWriterClose(SBseSnapWriter **pp, int8_t rollback) {
  int32_t code = 0;

  SBseSnapWriter *p = *pp;
  if (p == NULL) {
    return;
  }

  taosArrayDestroy(p->pFileSet);
  bseRawFileWriterClose(p->pWriter, 0);
  taosMemoryFree(p);
}

int32_t bseSnapReaderOpen(SBse *pBse, int64_t sver, int64_t ever, SBseSnapReader **ppReader) {
  int32_t code = 0;
  int32_t lino = 0;

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

  code = bseOpenIter(pBse, (SBseIter **)&p->pIter);
  TSDB_CHECK_CODE(code, lino, _error);

  p->pBse = pBse;

  *ppReader = p;
_error:
  if (code) {
    if (p != NULL) {
      bseError("vgId:%d failed to open table pReader at line %d since %s", BSE_VGID((SBse *)pBse), lino,
               tstrerror(code));
      bseSnapReaderClose(&p);
    }
    *ppReader = NULL;
  }
  return code;
}

int32_t bseSnapReaderReadImpl(SBseSnapReader *p, uint8_t **data, int32_t *len) {
  int32_t code = 0;
  int32_t line = 0;
  int32_t size = 0;

  uint8_t *pBuf = NULL;
  int32_t  bufLen = 0;

  if (bseIterIsOver(p->pIter)) {
    *data = NULL;
    return code;
  }

  code = bseIterNext(p->pIter, &pBuf, &bufLen);
  TSDB_CHECK_CODE(code, line, _error);

  if (bufLen == 0) {
    *data = NULL;
    return 0;
  }

  *data = taosMemoryCalloc(1, sizeof(SSnapDataHdr) + bufLen);
  if (*data == NULL) {
    TSDB_CHECK_CODE(code = terrno, line, _error);
  }

  SSnapDataHdr *pHdr = (SSnapDataHdr *)(*data);
  pHdr->type = SNAP_DATA_BSE;
  pHdr->size = bufLen;
  memcpy(pHdr->data, pBuf, bufLen);

  if (len != NULL) {
    *len = sizeof(SSnapDataHdr) + bufLen;
  }

_error:
  if (code) {
    if (*data != NULL) {
      taosMemoryFree(*data);
      *data = NULL;
    }
    bseError("vgId:%d failed to read snapshot data at line %d since %s", BSE_VGID((SBse *)p->pBse), line,
             tstrerror(code));
  }
  return code;
}
int32_t bseSnapReaderRead(SBseSnapReader *p, uint8_t **data) {
  int32_t len = 0;
  return bseSnapReaderReadImpl(p, data, NULL);
}
// test func
int32_t bseSnapReaderRead2(SBseSnapReader *p, uint8_t **data, int32_t *len) {
  return bseSnapReaderReadImpl(p, data, len);
}

void bseSnapReaderClose(SBseSnapReader **p) {
  int32_t code = 0;
  if (p == NULL || *p == NULL) {
    return;
  }

  SBseSnapReader *pReader = *p;
  bseIterDestroy(pReader->pIter);
  taosMemoryFree(pReader);

  *p = NULL;
}

int32_t bseOpenIter(SBse *pBse, SBseIter **ppIter) {
  int32_t code = 0;
  int32_t line = 0;

  SBseIter *pIter = taosMemoryCalloc(1, sizeof(SBseIter));
  if (pIter == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  SArray *pAliveFile = NULL;
  code = bseGetAliveFileList(pBse, &pAliveFile, 1);
  TSDB_CHECK_CODE(code, line, _error);

  pIter->index = 0;
  pIter->pFileSet = pAliveFile;
  pIter->fileType = BSE_TABLE_SNAP;
  pIter->pBse = pBse;

  *ppIter = pIter;

_error:
  if (code != 0) {
    bseError("vgId:%d failed to open iter since %s", BSE_VGID(pBse), tstrerror(code));
    taosMemoryFree(pIter);
    taosArrayDestroy(pAliveFile);
    return code;
  }
  return code;
}

int32_t bseIterNext(SBseIter *pIter, uint8_t **pValue, int32_t *len) {
  int32_t code = 0;
  int32_t lino = 0;

  STableReaderIter *pTableIter = NULL;

  if (pIter->fileType == BSE_TABLE_SNAP) {
    pTableIter = pIter->pTableIter;
    if (pTableIter != NULL && tableReaderIterValid(pTableIter)) {
      code = tableReaderIterNext(pTableIter, pValue, len);
      TSDB_CHECK_CODE(code, lino, _error);

      pTableIter->fileType = pIter->fileType;
      if (!tableReaderIterValid(pTableIter)) {
        // current file is over
        tableReaderIterDestroy(pTableIter);
        pIter->pTableIter = NULL;
      } else {
        return code;
      }
    }

    if (pIter->index >= taosArrayGetSize(pIter->pFileSet)) {
      pIter->fileType = BSE_TABLE_META_SNAP;
      pIter->index = 0;
    } else {
      if (pIter->pTableIter != NULL) {
        tableReaderIterDestroy(pIter->pTableIter);
        pIter->pTableIter = NULL;
      }

      SBseLiveFileInfo *pInfo = taosArrayGet(pIter->pFileSet, pIter->index);
      code = tableReaderIterInit(pInfo->timestamp, BSE_TABLE_DATA_TYPE, &pTableIter, pIter->pBse);
      TSDB_CHECK_CODE(code, lino, _error);

      pTableIter->fileType = BSE_TABLE_SNAP;
      code = tableReaderIterNext(pTableIter, pValue, len);
      TSDB_CHECK_CODE(code, lino, _error);

      pIter->pTableIter = pTableIter;

      pIter->index++;
      return code;
    }
  }

  if (pIter->fileType == BSE_TABLE_META_SNAP) {
    pTableIter = pIter->pTableIter;
    if (pTableIter != NULL && tableReaderIterValid(pTableIter)) {
      code = tableReaderIterNext(pTableIter, pValue, len);
      TSDB_CHECK_CODE(code, lino, _error);

      if (!tableReaderIterValid(pTableIter)) {
        tableReaderIterDestroy(pTableIter);
        pIter->pTableIter = NULL;
      } else {
        return code;
      }
    }
    if (pIter->index >= taosArrayGetSize(pIter->pFileSet)) {
      pIter->fileType = BSE_CURRENT_SNAP;
    } else {
      if (pIter->pTableIter != NULL) {
        tableReaderIterDestroy(pIter->pTableIter);
        pIter->pTableIter = NULL;
      }

      SBseLiveFileInfo *pInfo = taosArrayGet(pIter->pFileSet, pIter->index);
      code = tableReaderIterInit(pInfo->timestamp, BSE_TABLE_META_TYPE, &pTableIter, pIter->pBse);
      TSDB_CHECK_CODE(code, lino, _error);

      code = tableReaderIterNext(pTableIter, pValue, len);
      TSDB_CHECK_CODE(code, lino, _error);

      pIter->pTableIter = pTableIter;
      pIter->index++;
      return code;
    }
  }

  if (pIter->fileType == BSE_CURRENT_SNAP) {
    code = bseReadCurrentSnap(pIter->pBse, pValue, len);
    // do read current
    pIter->fileType = BSE_MAX_SNAP;
    pIter->isOver = 1;

    pIter->pCurrentBuf = *pValue;
  } else if (pIter->fileType == BSE_MAX_SNAP) {
    pIter->isOver = 1;
  }

_error:
  if (code != 0) {
    bseError("vgId:%d failed to get next iter since %s", BSE_VGID(pIter->pBse), tstrerror(code));
  }
  return code;
}

void bseIterDestroy(SBseIter *pIter) {
  if (pIter == NULL) {
    return;
  }

  if (pIter->pTableIter != NULL) {
    tableReaderIterDestroy(pIter->pTableIter);
  }

  taosArrayDestroy(pIter->pFileSet);
  if (pIter->pCurrentBuf != NULL) {
    taosMemoryFreeClear(pIter->pCurrentBuf);
  }
  taosMemFree(pIter);
  return;
}
int8_t bseIterValid(SBseIter *pIter) {
  if (pIter == NULL) {
    return 0;
  }
  return pIter->isOver == 0;
}
int8_t bseIterIsOver(SBseIter *pIter) {
  if (pIter == NULL) {
    return 1;
  }
  return pIter->isOver == 1;
}

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 "bseSnapshot.h"
17	#include "bseInc.h"
18	#include "bseTable.h"
19	#include "bseTableMgt.h"
20	#include "bseUtil.h"
21	#include "vnodeInt.h"
22
23	static int32_t bseRawFileWriterOpen(SBse pBse, int64_t sver, int64_t ever, SBseSnapMeta pMeta,
24	SBseRawFileWriter **pWriter);
25	static int32_t bseRawFileWriterDoWrite(SBseRawFileWriter p, uint8_t data, int32_t len);
26	static void bseRawFileWriterClose(SBseRawFileWriter *p, int8_t rollback);
27	static void bseRawFileGenLiveInfo(SBseRawFileWriter p, SBseLiveFileInfo pInfo);
28
29	static int32_t bseSnapReaderReadImpl(SBseSnapReader p, uint8_t data, int32_t len);
30
31	static int32_t bseRawFileWriterOpen(SBse pBse, int64_t sver, int64_t ever, SBseSnapMeta pMeta,	×
32	SBseRawFileWriter **pWriter) {
33	int32_t code = 0;	×
34	int32_t lino = 0;	×
35
36	char name[TSDB_FILENAME_LEN] = {0};	×
37	char path[TSDB_FILENAME_LEN] = {0};	×
38
39	SBseRawFileWriter *p = taosMemoryCalloc(1, sizeof(SBseRawFileWriter));	×
40	if (p == NULL) {	×
41	TSDB_CHECK_CODE(code = terrno, lino, _error);	×
42	}
43
44	SSeqRange *range = &pMeta->range;	×
45	if (pMeta->fileType == BSE_TABLE_SNAP) {	×
46	bseBuildDataName(pMeta->timestamp, name);	×
47	bseBuildFullName(pBse, name, path);	×
48	} else if (pMeta->fileType == BSE_TABLE_META_TYPE) {	×
49	bseBuildMetaName(pMeta->timestamp, name);	×
50	bseBuildFullName(pBse, name, path);	×
51	} else if (pMeta->fileType == BSE_CURRENT_SNAP) {	×
52	bseBuildCurrentFullName(pBse, path);	×
53	} else {
54	return TSDB_CODE_INVALID_MSG;	×
55	}
56
57	p->fileType = pMeta->fileType;	×
58	p->range = *range;	×
59	p->pBse = pBse;	×
60	p->pFile = taosOpenFile(path, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_READ \| TD_FILE_APPEND \| TD_FILE_TRUNC);	×
61	if (p->pFile == NULL) {	×
62	TSDB_CHECK_CODE(code = terrno, lino, _error);	×
63	}
64
65	*pWriter = p;	×
66	_error:	×
67	if (code) {	×
68	if (p != NULL) {	×
69	bseError("vgId:%d failed to open table pWriter at line %d since %s", BSE_VGID((SBse *)pBse), lino,	×
70	tstrerror(code));
71	bseRawFileWriterClose(p, 0);	×
72	}
73	*pWriter = NULL;	×
74	}
75
76	return code;	×
77	}
78
79	int32_t bseRawFileWriterDoWrite(SBseRawFileWriter p, uint8_t data, int32_t len) {	×
80	int32_t code = 0;	×
81	int32_t nwrite = taosWriteFile(p->pFile, data, len);	×
82	if (nwrite != len) {	×
83	return terrno;	×
84	}
85	p->offset += len;	×
86
87	return code;	×
88	}
89
90	void bseRawFileWriterClose(SBseRawFileWriter *p, int8_t rollback) {	×
91	if (p == NULL) return;	×
92
93	int32_t code = 0;	×
94	if (taosCloseFile(&p->pFile) != 0) {	×
95	bseError("vgId:%d failed to close table pWriter since %s", BSE_VGID((SBse *)p->pBse), tstrerror(terrno));	×
96	}
97	taosMemoryFree(p);	×
98
99	return;	×
100	}
101
102	void bseRawFileGenLiveInfo(SBseRawFileWriter p, SBseLiveFileInfo pInfo) {	×
103	pInfo->range = p->range;	×
104	pInfo->size = p->offset;	×
105	memcpy(pInfo->name, p->name, sizeof(p->name));	×
106	}	×
107
108	int32_t bseSnapMayOpenNewFile(SBseSnapWriter pWriter, SBseSnapMeta pMeta) {	×
109	int32_t code = 0;	×
110
111	SBse *pBse = pWriter->pBse;	×
112
113	SBseRawFileWriter *pOld = pWriter->pWriter;	×
114
115	if (pOld == NULL \|\| (pOld->fileType != pMeta->fileType)) {	×
116	if (pOld != NULL) {	×
UNCOV 117	SBseLiveFileInfo info;	×
118	bseRawFileGenLiveInfo(pOld, &info);	×
119
120	if (taosArrayPush(pWriter->pFileSet, &info) == NULL) {	×
121	code = terrno;	×
122	bseError("vgId:%d failed to push file info since %s", BSE_VGID((SBse *)pBse), tstrerror(code));	×
123	return code;	×
124	}
125	bseRawFileWriterClose(pOld, 0);	×
126	pWriter->pWriter = NULL;	×
127	}
128
129	SBseRawFileWriter *pNew = NULL;	×
130	code = bseRawFileWriterOpen(pBse, 0, 0, pMeta, &pNew);	×
131	if (code) {	×
132	bseError("vgId:%d failed to open table pWriter since %s", BSE_VGID((SBse *)pBse), tstrerror(code));	×
133	return code;	×
134	}
135	pWriter->pWriter = pNew;	×
136	}
137	return code;	×
138	}
139
140	int32_t bseSnapWriterOpen(SBse pBse, int64_t sver, int64_t ever, SBseSnapWriter *pWriter) {	×
141	int32_t code = 0;	×
142	int32_t lino = 0;	×
143
144	SBseSnapWriter *p = taosMemoryCalloc(1, sizeof(SBseSnapWriter));	×
145	if (p == NULL) {	×
146	TSDB_CHECK_CODE(code = terrno, lino, _error);	×
147	}
148
149	p->pFileSet = taosArrayInit(128, sizeof(SBseLiveFileInfo));	×
150	if (p->pFileSet == NULL) {	×
151	TSDB_CHECK_CODE(code = terrno, lino, _error);	×
152	}
153	p->pBse = pBse;	×
154
155	*pWriter = p;	×
156
157	_error:	×
158	if (code) {	×
159	if (p != NULL) {	×
160	bseError("vgId:%d failed to open table pWriter at line %d since %s", BSE_VGID((SBse *)pBse), lino,	×
161	tstrerror(code));
162	bseSnapWriterClose(&p, 0);	×
163	}
164	*pWriter = NULL;	×
165	}
166
167	return code;	×
168	}
169	int32_t bseSnapWriterWrite(SBseSnapWriter p, uint8_t data, int32_t len) {	×
170	int32_t code;
171	int32_t lino = 0;	×
172	SSnapDataHdr pHdr = (SSnapDataHdr )data;	×
173	if (pHdr->size + sizeof(SSnapDataHdr) != len) {	×
174	return TSDB_CODE_INVALID_MSG;	×
175	}
176	SBseSnapMeta pMeta = (SBseSnapMeta )pHdr->data;	×
177
178	uint8_t *pBuf = pHdr->data + sizeof(SBseSnapMeta);	×
179	int64_t tlen = len - sizeof(SSnapDataHdr) - sizeof(SBseSnapMeta);	×
180
181	code = bseSnapMayOpenNewFile(p, pMeta);	×
182	TSDB_CHECK_CODE(code, lino, _error);	×
183
184	code = bseRawFileWriterDoWrite(p->pWriter, pBuf, tlen);	×
185	TSDB_CHECK_CODE(code, lino, _error);	×
186	_error:	×
187	if (code) {	×
188	if (p->pWriter != NULL) {	×
189	bseError("vgId:%d failed to write snapshot data since %s", BSE_VGID((SBse *)p->pBse), tstrerror(code));	×
190	bseRawFileWriterClose(p->pWriter, 0);	×
191	}
192	return code;	×
193	}
194	return code;	×
195	}
196	void bseSnapWriterClose(SBseSnapWriter **pp, int8_t rollback) {	×
197	int32_t code = 0;	×
198
199	SBseSnapWriter p = pp;	×
200	if (p == NULL) {	×
201	return;	×
202	}
203
204	taosArrayDestroy(p->pFileSet);	×
205	bseRawFileWriterClose(p->pWriter, 0);	×
206	taosMemoryFree(p);	×
207	}
208
209	int32_t bseSnapReaderOpen(SBse pBse, int64_t sver, int64_t ever, SBseSnapReader *ppReader) {	×
210	int32_t code = 0;	×
211	int32_t lino = 0;	×
212
213	SBseSnapReader *p = taosMemoryCalloc(1, sizeof(SBseSnapReader));	×
214	if (p == NULL) {	×
215	TSDB_CHECK_CODE(code = terrno, lino, _error);	×
216	}
217
218	code = bseOpenIter(pBse, (SBseIter **)&p->pIter);	×
219	TSDB_CHECK_CODE(code, lino, _error);	×
220
221	p->pBse = pBse;	×
222
223	*ppReader = p;	×
224	_error:	×
225	if (code) {	×
226	if (p != NULL) {	×
227	bseError("vgId:%d failed to open table pReader at line %d since %s", BSE_VGID((SBse *)pBse), lino,	×
228	tstrerror(code));
229	bseSnapReaderClose(&p);	×
230	}
231	*ppReader = NULL;	×
232	}
233	return code;	×
234	}
235
236	int32_t bseSnapReaderReadImpl(SBseSnapReader p, uint8_t data, int32_t len) {	×
237	int32_t code = 0;	×
238	int32_t line = 0;	×
239	int32_t size = 0;	×
240
241	uint8_t *pBuf = NULL;	×
242	int32_t bufLen = 0;	×
243
244	if (bseIterIsOver(p->pIter)) {	×
245	*data = NULL;	×
246	return code;	×
247	}
248
249	code = bseIterNext(p->pIter, &pBuf, &bufLen);	×
250	TSDB_CHECK_CODE(code, line, _error);	×
251
252	if (bufLen == 0) {	×
253	*data = NULL;	×
254	return 0;	×
255	}
256
257	*data = taosMemoryCalloc(1, sizeof(SSnapDataHdr) + bufLen);	×
258	if (*data == NULL) {	×
259	TSDB_CHECK_CODE(code = terrno, line, _error);	×
260	}
261
262	SSnapDataHdr pHdr = (SSnapDataHdr )(*data);	×
263	pHdr->type = SNAP_DATA_BSE;	×
264	pHdr->size = bufLen;	×
265	memcpy(pHdr->data, pBuf, bufLen);	×
266
267	if (len != NULL) {	×
268	*len = sizeof(SSnapDataHdr) + bufLen;	×
269	}
270
271	_error:	×
272	if (code) {	×
273	if (*data != NULL) {	×
274	taosMemoryFree(*data);	×
275	*data = NULL;	×
276	}
277	bseError("vgId:%d failed to read snapshot data at line %d since %s", BSE_VGID((SBse *)p->pBse), line,	×
278	tstrerror(code));
279	}
280	return code;	×
281	}
282	int32_t bseSnapReaderRead(SBseSnapReader p, uint8_t *data) {	×
283	int32_t len = 0;	×
284	return bseSnapReaderReadImpl(p, data, NULL);	×
285	}
286	// test func
287	int32_t bseSnapReaderRead2(SBseSnapReader p, uint8_t data, int32_t len) {	×
288	return bseSnapReaderReadImpl(p, data, len);	×
289	}
290
291	void bseSnapReaderClose(SBseSnapReader **p) {	×
292	int32_t code = 0;	×
293	if (p == NULL \|\| *p == NULL) {	×
294	return;	×
295	}
296
297	SBseSnapReader pReader = p;	×
298	bseIterDestroy(pReader->pIter);	×
299	taosMemoryFree(pReader);	×
300
301	*p = NULL;	×
302	}
303
304	int32_t bseOpenIter(SBse pBse, SBseIter *ppIter) {	×
305	int32_t code = 0;	×
306	int32_t line = 0;	×
307
308	SBseIter *pIter = taosMemoryCalloc(1, sizeof(SBseIter));	×
309	if (pIter == NULL) {	×
310	return TSDB_CODE_OUT_OF_MEMORY;	×
311	}
312
313	SArray *pAliveFile = NULL;	×
314	code = bseGetAliveFileList(pBse, &pAliveFile, 1);	×
315	TSDB_CHECK_CODE(code, line, _error);	×
316
317	pIter->index = 0;	×
318	pIter->pFileSet = pAliveFile;	×
319	pIter->fileType = BSE_TABLE_SNAP;	×
320	pIter->pBse = pBse;	×
321
322	*ppIter = pIter;	×
323
324	_error:	×
325	if (code != 0) {	×
326	bseError("vgId:%d failed to open iter since %s", BSE_VGID(pBse), tstrerror(code));	×
327	taosMemoryFree(pIter);	×
328	taosArrayDestroy(pAliveFile);	×
329	return code;	×
330	}
331	return code;	×
332	}
333
334	int32_t bseIterNext(SBseIter pIter, uint8_t pValue, int32_t len) {	×
335	int32_t code = 0;	×
336	int32_t lino = 0;	×
337
338	STableReaderIter *pTableIter = NULL;	×
339
340	if (pIter->fileType == BSE_TABLE_SNAP) {	×
341	pTableIter = pIter->pTableIter;	×
342	if (pTableIter != NULL && tableReaderIterValid(pTableIter)) {	×
343	code = tableReaderIterNext(pTableIter, pValue, len);	×
344	TSDB_CHECK_CODE(code, lino, _error);	×
345
346	pTableIter->fileType = pIter->fileType;	×
347	if (!tableReaderIterValid(pTableIter)) {	×
348	// current file is over
349	tableReaderIterDestroy(pTableIter);	×
350	pIter->pTableIter = NULL;	×
351	} else {
352	return code;	×
353	}
354	}
355
356	if (pIter->index >= taosArrayGetSize(pIter->pFileSet)) {	×
357	pIter->fileType = BSE_TABLE_META_SNAP;	×
358	pIter->index = 0;	×
359	} else {
360	if (pIter->pTableIter != NULL) {	×
361	tableReaderIterDestroy(pIter->pTableIter);	×
362	pIter->pTableIter = NULL;	×
363	}
364
365	SBseLiveFileInfo *pInfo = taosArrayGet(pIter->pFileSet, pIter->index);	×
366	code = tableReaderIterInit(pInfo->timestamp, BSE_TABLE_DATA_TYPE, &pTableIter, pIter->pBse);	×
367	TSDB_CHECK_CODE(code, lino, _error);	×
368
369	pTableIter->fileType = BSE_TABLE_SNAP;	×
370	code = tableReaderIterNext(pTableIter, pValue, len);	×
371	TSDB_CHECK_CODE(code, lino, _error);	×
372
373	pIter->pTableIter = pTableIter;	×
374
375	pIter->index++;	×
376	return code;	×
377	}
378	}
379
380	if (pIter->fileType == BSE_TABLE_META_SNAP) {	×
381	pTableIter = pIter->pTableIter;	×
382	if (pTableIter != NULL && tableReaderIterValid(pTableIter)) {	×
383	code = tableReaderIterNext(pTableIter, pValue, len);	×
384	TSDB_CHECK_CODE(code, lino, _error);	×
385
386	if (!tableReaderIterValid(pTableIter)) {	×
387	tableReaderIterDestroy(pTableIter);	×
388	pIter->pTableIter = NULL;	×
389	} else {
390	return code;	×
391	}
392	}
393	if (pIter->index >= taosArrayGetSize(pIter->pFileSet)) {	×
394	pIter->fileType = BSE_CURRENT_SNAP;	×
395	} else {
396	if (pIter->pTableIter != NULL) {	×
397	tableReaderIterDestroy(pIter->pTableIter);	×
398	pIter->pTableIter = NULL;	×
399	}
400
401	SBseLiveFileInfo *pInfo = taosArrayGet(pIter->pFileSet, pIter->index);	×
402	code = tableReaderIterInit(pInfo->timestamp, BSE_TABLE_META_TYPE, &pTableIter, pIter->pBse);	×
403	TSDB_CHECK_CODE(code, lino, _error);	×
404
405	code = tableReaderIterNext(pTableIter, pValue, len);	×
406	TSDB_CHECK_CODE(code, lino, _error);	×
407
408	pIter->pTableIter = pTableIter;	×
409	pIter->index++;	×
410	return code;	×
411	}
412	}
413
414	if (pIter->fileType == BSE_CURRENT_SNAP) {	×
415	code = bseReadCurrentSnap(pIter->pBse, pValue, len);	×
416	// do read current
417	pIter->fileType = BSE_MAX_SNAP;	×
418	pIter->isOver = 1;	×
419
420	pIter->pCurrentBuf = *pValue;	×
421	} else if (pIter->fileType == BSE_MAX_SNAP) {	×
422	pIter->isOver = 1;	×
423	}
424
425	_error:	×
426	if (code != 0) {	×
427	bseError("vgId:%d failed to get next iter since %s", BSE_VGID(pIter->pBse), tstrerror(code));	×
428	}
429	return code;	×
430	}
431
432	void bseIterDestroy(SBseIter *pIter) {	×
433	if (pIter == NULL) {	×
434	return;	×
435	}
436
437	if (pIter->pTableIter != NULL) {	×
438	tableReaderIterDestroy(pIter->pTableIter);	×
439	}
440
441	taosArrayDestroy(pIter->pFileSet);	×
442	if (pIter->pCurrentBuf != NULL) {	×
443	taosMemoryFreeClear(pIter->pCurrentBuf);	×
444	}
445	taosMemFree(pIter);	×
446	return;	×
447	}
448	int8_t bseIterValid(SBseIter *pIter) {	×
449	if (pIter == NULL) {	×
450	return 0;	×
451	}
452	return pIter->isOver == 0;	×
453	}
454	int8_t bseIterIsOver(SBseIter *pIter) {	×
455	if (pIter == NULL) {	×
456	return 1;	×
457	}
458	return pIter->isOver == 1;	×
459	}

taosdata / TDengine / #4987

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