#3653

Committed 14 Mar 2025 08:10AM UTC coverage: 22.565% (-41.0%) from 63.596%

Build # #3653

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(keep): support keep on super table level. (#30097)

* Feat: support use keep while create super table.

* Test(keep): add test for create super table with keep option.

* Feat(keep): Add tmsg for create keep.

* Feat(keep): support alter table option keep.

* Fix(keep): Add baisc test for alter table option.

* Fix(keep): memory leek.

* Feat(keep): add keep to metaEntry&metaCache and fix earliestTs with stn keep.

* Test(keep): add some cases for select with stb keep.

* Fix: fix ci core while alter stb.

* Feat(keep): delete expired data in super table level.

* Feat: remove get stb keep while query.

* Fix : build error.

* Revert "Fix : build error."

This reverts commit 0ed66e4e8.

* Revert "Feat(keep): delete expired data in super table level."

This reverts commit 36330f6b4.

* Fix : build errors.

* Feat : support restart taosd.

* Fix : alter table comment problems.

* Test : add tests for super table keep.

* Fix: change sdb stb reserve size.

* Test: add more tests.

* Feat: Disable normal tables and sub tables from setting the keep parameter

* Fix: add more checks to avoid unknown address.

* Docs: Add docs for stable keep.

* Fix: some review changes.

* Fix: review errors.

Run Details

49248 of 302527 branches covered (16.28%)

Branch coverage included in aggregate %.

53 of 99 new or added lines in 12 files covered. (53.54%)

155872 existing lines in 443 files now uncovered.

87359 of 302857 relevant lines covered (28.84%)

570004.22 hits per line

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

14.58

/source/libs/sync/src/syncCommit.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/>.
 */

#define _DEFAULT_SOURCE
#include "syncCommit.h"
#include "syncIndexMgr.h"
#include "syncRaftLog.h"
#include "syncRaftStore.h"
#include "syncUtil.h"

// \* Leader i advances its commitIndex.
// \* This is done as a separate step from handling AppendEntries responses,
// \* in part to minimize atomic regions, and in part so that leaders of
// \* single-server clusters are able to mark entries committed.
// AdvanceCommitIndex(i) ==
//     /\ state[i] = Leader
//     /\ LET \* The set of servers that agree up through index.
//            Agree(index) == {i} \cup {k \in Server :
//                                          matchIndex[i][k] >= index}
//            \* The maximum indexes for which a quorum agrees
//            agreeIndexes == {index \in 1..Len(log[i]) :
//                                 Agree(index) \in Quorum}
//            \* New value for commitIndex'[i]
//            newCommitIndex ==
//               IF /\ agreeIndexes /= {}
//                  /\ log[i][Max(agreeIndexes)].term = currentTerm[i]
//               THEN
//                   Max(agreeIndexes)
//               ELSE
//                   commitIndex[i]
//        IN commitIndex' = [commitIndex EXCEPT ![i] = newCommitIndex]
//     /\ UNCHANGED <<messages, serverVars, candidateVars, leaderVars, log>>
//

static inline int64_t syncNodeAbs64(int64_t a, int64_t b) {
  int64_t c = a > b ? a - b : b - a;
  return c;
}

int32_t syncNodeDynamicQuorum(const SSyncNode* pSyncNode) { return pSyncNode->quorum; }

bool syncNodeAgreedUpon(SSyncNode* pNode, SyncIndex index) {
  int            count = 0;
  SSyncIndexMgr* pMatches = pNode->pMatchIndex;
  if (pNode->replicaNum != pMatches->replicaNum) {
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return false;
  };

  for (int i = 0; i < pNode->totalReplicaNum; i++) {
    if(pNode->raftCfg.cfg.nodeInfo[i].nodeRole == TAOS_SYNC_ROLE_VOTER){
      SyncIndex matchIndex = pMatches->index[i];
      if (matchIndex >= index) {
        count++;
      }
    }
  }

  return count >= pNode->quorum;
}

int64_t syncNodeUpdateCommitIndex(SSyncNode* ths, SyncIndex commitIndex) {
  int32_t   code = 0;
  SyncIndex lastVer = ths->pLogStore->syncLogLastIndex(ths->pLogStore);
  commitIndex = TMAX(commitIndex, ths->commitIndex);
  ths->commitIndex = TMIN(commitIndex, lastVer);
  // TODO add return when error
  (void)ths->pLogStore->syncLogUpdateCommitIndex(ths->pLogStore, ths->commitIndex);
  return ths->commitIndex;
}

int64_t syncNodeCheckCommitIndex(SSyncNode* ths, SyncIndex indexLikely) {
  int32_t code = 0;
  if (indexLikely > ths->commitIndex && syncNodeAgreedUpon(ths, indexLikely)) {
    SyncIndex commitIndex = indexLikely;
    SyncIndex returnIndex = syncNodeUpdateCommitIndex(ths, commitIndex);
    sTrace("vgId:%d, agreed upon. role:%d, term:%" PRId64 ", index:%" PRId64 ", return:%" PRId64, ths->vgId, ths->state,
           raftStoreGetTerm(ths), commitIndex, returnIndex);
  }
  return ths->commitIndex;
}

int64_t syncNodeUpdateAssignedCommitIndex(SSyncNode* ths, SyncIndex assignedCommitIndex) {
  int32_t   code = 0;
  SyncIndex lastVer = ths->pLogStore->syncLogLastIndex(ths->pLogStore);
  assignedCommitIndex = TMAX(assignedCommitIndex, ths->assignedCommitIndex);
  ths->assignedCommitIndex = TMIN(assignedCommitIndex, lastVer);
  // TODO add return when error
  (void)ths->pLogStore->syncLogUpdateCommitIndex(ths->pLogStore, ths->assignedCommitIndex);
  return ths->commitIndex;
}

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	#define _DEFAULT_SOURCE
17	#include "syncCommit.h"
18	#include "syncIndexMgr.h"
19	#include "syncRaftLog.h"
20	#include "syncRaftStore.h"
21	#include "syncUtil.h"
22
23	// \* Leader i advances its commitIndex.
24	// \* This is done as a separate step from handling AppendEntries responses,
25	// \* in part to minimize atomic regions, and in part so that leaders of
26	// \* single-server clusters are able to mark entries committed.
27	// AdvanceCommitIndex(i) ==
28	// /\ state[i] = Leader
29	// /\ LET \* The set of servers that agree up through index.
30	// Agree(index) == {i} \cup {k \in Server :
31	// matchIndex[i][k] >= index}
32	// \* The maximum indexes for which a quorum agrees
33	// agreeIndexes == {index \in 1..Len(log[i]) :
34	// Agree(index) \in Quorum}
35	// \* New value for commitIndex'[i]
36	// newCommitIndex ==
37	// IF /\ agreeIndexes /= {}
38	// /\ log[i][Max(agreeIndexes)].term = currentTerm[i]
39	// THEN
40	// Max(agreeIndexes)
41	// ELSE
42	// commitIndex[i]
43	// IN commitIndex' = [commitIndex EXCEPT ![i] = newCommitIndex]
44	// /\ UNCHANGED <<messages, serverVars, candidateVars, leaderVars, log>>
45	//
46
47	static inline int64_t syncNodeAbs64(int64_t a, int64_t b) {
48	int64_t c = a > b ? a - b : b - a;
49	return c;
50	}
51
UNCOV 52	int32_t syncNodeDynamicQuorum(const SSyncNode* pSyncNode) { return pSyncNode->quorum; }	×
53
UNCOV 54	bool syncNodeAgreedUpon(SSyncNode* pNode, SyncIndex index) {	×
UNCOV 55	int count = 0;	×
UNCOV 56	SSyncIndexMgr* pMatches = pNode->pMatchIndex;	×
UNCOV 57	if (pNode->replicaNum != pMatches->replicaNum) {	×
58	terrno = TSDB_CODE_SYN_INTERNAL_ERROR;	×
59	return false;	×
60	};
61
UNCOV 62	for (int i = 0; i < pNode->totalReplicaNum; i++) {	×
UNCOV 63	if(pNode->raftCfg.cfg.nodeInfo[i].nodeRole == TAOS_SYNC_ROLE_VOTER){	×
UNCOV 64	SyncIndex matchIndex = pMatches->index[i];	×
UNCOV 65	if (matchIndex >= index) {	×
UNCOV 66	count++;	×
67	}
68	}
69	}
70
UNCOV 71	return count >= pNode->quorum;	×
72	}
73
74	int64_t syncNodeUpdateCommitIndex(SSyncNode* ths, SyncIndex commitIndex) {	298✔
75	int32_t code = 0;	298✔
76	SyncIndex lastVer = ths->pLogStore->syncLogLastIndex(ths->pLogStore);	298✔
77	commitIndex = TMAX(commitIndex, ths->commitIndex);	298✔
78	ths->commitIndex = TMIN(commitIndex, lastVer);	298✔
79	// TODO add return when error
80	(void)ths->pLogStore->syncLogUpdateCommitIndex(ths->pLogStore, ths->commitIndex);	298✔
81	return ths->commitIndex;	298✔
82	}
83
UNCOV 84	int64_t syncNodeCheckCommitIndex(SSyncNode* ths, SyncIndex indexLikely) {	×
UNCOV 85	int32_t code = 0;	×
UNCOV 86	if (indexLikely > ths->commitIndex && syncNodeAgreedUpon(ths, indexLikely)) {	×
UNCOV 87	SyncIndex commitIndex = indexLikely;	×
UNCOV 88	SyncIndex returnIndex = syncNodeUpdateCommitIndex(ths, commitIndex);	×
UNCOV 89	sTrace("vgId:%d, agreed upon. role:%d, term:%" PRId64 ", index:%" PRId64 ", return:%" PRId64, ths->vgId, ths->state,	×
90	raftStoreGetTerm(ths), commitIndex, returnIndex);
91	}
UNCOV 92	return ths->commitIndex;	×
93	}
94
95	int64_t syncNodeUpdateAssignedCommitIndex(SSyncNode* ths, SyncIndex assignedCommitIndex) {	×
96	int32_t code = 0;	×
97	SyncIndex lastVer = ths->pLogStore->syncLogLastIndex(ths->pLogStore);	×
98	assignedCommitIndex = TMAX(assignedCommitIndex, ths->assignedCommitIndex);	×
99	ths->assignedCommitIndex = TMIN(assignedCommitIndex, lastVer);	×
100	// TODO add return when error
101	(void)ths->pLogStore->syncLogUpdateCommitIndex(ths->pLogStore, ths->assignedCommitIndex);	×
102	return ths->commitIndex;	×
103	}

taosdata / TDengine / #3653

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