Skip to content

Commit 7606fce

Browse files
luwei16yiguolei
luwei16
authored and
yiguolei
committed
branch-4.1: [fix](cloud) Invalidate version caches on visible commit retries (#67813) (#68650)
### What problem does this PR solve? Issue Number: #67099 Related PR: #67813 Problem Summary: A retry after a lost cloud commit response can return a VISIBLE transaction without version results. Invalidate the affected table and partition version caches, and notify other FEs so subsequent reads fetch current versions from the MetaService. This cherry-picks the merged #67813 commit (`3dbe4d3ca53307f73db96eda23d4efc9f02963c5`) to `branch-4.1`. The conflict resolution keeps the branch-4.1 partition cache API, which has no commit-TSO argument, and adapts the tests to its Java 8 and JUnit 4 setup. It preserves the source PR's cache invalidation epochs and batch publication locks. ### Release note Prevent stale reads and false commit failures after cloud commit retries. ### Check List (For Author) - Test - [x] Unit Test: 51 passed (`CloudGlobalTransactionMgrTest`, `CloudPartitionTest`, `OlapTableTest`, `VersionHelperTest`). - [x] FE Checkstyle: 0 violations; `git diff --check` passed. - [ ] Live cluster / SQL regression: not run. - Behavior changed: - [x] Yes. VISIBLE retries without versions invalidate local and peer caches; subsequent reads fetch fresh versions without changing the successful commit outcome. - Does this need documentation? - [x] No.
1 parent a7daed6 commit 7606fce

5 files changed

Lines changed: 520 additions & 13 deletions

File tree

‎fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -242,6 +242,10 @@ public enum OlapTableState {
242242
private volatile long lastTableVersionCachedTimeMs = 0;
243243
private volatile long cachedTableVersion = -1;
244244

245+
// Commit notifications cannot clear invalidation: their versions may precede the missing commit result.
246+
private final AtomicLong tableVersionCacheEpoch = new AtomicLong();
247+
private final AtomicLong refreshedTableVersionCacheEpoch = new AtomicLong();
248+
245249
private ReadWriteLock versionLock = Config.isCloudMode() ? new ReentrantReadWriteLock(true) : null;
246250

247251
public OlapTable() {
@@ -3369,7 +3373,7 @@ public long getNextVersion() {
33693373
@VisibleForTesting
33703374
protected boolean isCachedTableVersionExpired() {
33713375
// -1 means no cache yet, need to fetch from MS
3372-
if (cachedTableVersion == -1) {
3376+
if (cachedTableVersion == -1 || tableVersionCacheEpoch.get() != refreshedTableVersionCacheEpoch.get()) {
33733377
return true;
33743378
}
33753379
ConnectContext ctx = ConnectContext.get();
@@ -3383,13 +3387,18 @@ protected boolean isCachedTableVersionExpired() {
33833387

33843388
public boolean isCachedTableVersionExpired(long expirationMs) {
33853389
// -1 means no cache yet, need to fetch from MS
3386-
if (cachedTableVersion == -1 || expirationMs <= 0) {
3390+
if (cachedTableVersion == -1 || expirationMs <= 0
3391+
|| tableVersionCacheEpoch.get() != refreshedTableVersionCacheEpoch.get()) {
33873392
return true;
33883393
}
33893394
return System.currentTimeMillis() - lastTableVersionCachedTimeMs > expirationMs;
33903395
}
33913396

3392-
public void setCachedTableVersion(long version) {
3397+
public void invalidateCachedTableVersion() {
3398+
tableVersionCacheEpoch.incrementAndGet();
3399+
}
3400+
3401+
public synchronized void setCachedTableVersion(long version) {
33933402
if (version >= cachedTableVersion) {
33943403
cachedTableVersion = version;
33953404
lastTableVersionCachedTimeMs = System.currentTimeMillis();
@@ -3410,6 +3419,7 @@ public long getVisibleVersion() throws RpcException {
34103419
return getCachedTableVersion();
34113420
}
34123421

3422+
long cacheEpoch = tableVersionCacheEpoch.get();
34133423
// get version rpc
34143424
Cloud.GetVersionRequest request = Cloud.GetVersionRequest.newBuilder()
34153425
.setRequestIp(FrontendOptions.getLocalHostAddressCached())
@@ -3436,6 +3446,7 @@ public long getVisibleVersion() throws RpcException {
34363446
}
34373447
// update cache
34383448
setCachedTableVersion(version);
3449+
refreshedTableVersionCacheEpoch.accumulateAndGet(cacheEpoch, Math::max);
34393450
return version;
34403451
} catch (RpcException e) {
34413452
LOG.warn("get version from meta service failed", e);
@@ -3500,9 +3511,11 @@ public static List<Long> getVisibleVersionInBatch(List<OlapTable> tables) {
35003511
private static List<Long> getVisibleVersionInBatchFromMs(List<OlapTable> tables) {
35013512
List<Long> dbIds = new ArrayList<>(tables.size());
35023513
List<Long> tableIds = new ArrayList<>(tables.size());
3514+
List<Long> cacheEpochs = new ArrayList<>(tables.size());
35033515
for (OlapTable table : tables) {
35043516
dbIds.add(table.getDatabase().getId());
35053517
tableIds.add(table.getId());
3518+
cacheEpochs.add(table.tableVersionCacheEpoch.get());
35063519
}
35073520

35083521
List<Long> versions = getVisibleVersionFromMeta(dbIds, tableIds);
@@ -3511,6 +3524,7 @@ private static List<Long> getVisibleVersionInBatchFromMs(List<OlapTable> tables)
35113524
Preconditions.checkState(tables.size() == versions.size());
35123525
for (int i = 0; i < tables.size(); i++) {
35133526
tables.get(i).setCachedTableVersion(versions.get(i));
3527+
tables.get(i).refreshedTableVersionCacheEpoch.accumulateAndGet(cacheEpochs.get(i), Math::max);
35143528
}
35153529
return versions;
35163530
}

‎fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudFEVersionSynchronizer.java‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,9 @@
4949

5050
public class CloudFEVersionSynchronizer {
5151
private static final Logger LOG = LogManager.getLogger(CloudFEVersionSynchronizer.class);
52+
// Reuse the version RPC for whole-table invalidation when a commit response has no versions.
53+
// Older FEs ignore this value and continue relying on their periodic version synchronization.
54+
private static final long INVALIDATE_VERSION_CACHE = -1;
5255

5356
private static final ExecutorService SYNC_VERSION_THREAD_POOL = Executors.newFixedThreadPool(
5457
Config.cloud_sync_version_task_threads_num,
@@ -57,6 +60,54 @@ public class CloudFEVersionSynchronizer {
5760
public CloudFEVersionSynchronizer() {
5861
}
5962

63+
public void invalidateVersionCaches(long dbId, List<Long> tableIds) {
64+
Database db = Env.getCurrentInternalCatalog().getDbNullable(dbId);
65+
if (db == null) {
66+
return; // The database may have been dropped since the original commit.
67+
}
68+
List<OlapTable> tables = new ArrayList<>();
69+
for (long tableId : tableIds) {
70+
Table table = db.getTableNullable(tableId);
71+
if (table != null && table.isManagedTable()) {
72+
tables.add((OlapTable) table);
73+
}
74+
}
75+
invalidateVersionCaches(tables);
76+
List<Pair<OlapTable, Long>> tableVersions = tables.stream()
77+
.map(table -> Pair.of(table, INVALIDATE_VERSION_CACHE)).collect(Collectors.toList());
78+
pushVersionAsync(dbId, tableVersions, Collections.emptyMap());
79+
}
80+
81+
private void invalidateVersionCaches(List<OlapTable> tables) {
82+
tables.sort(Comparator.comparingLong(OlapTable::getId));
83+
for (OlapTable table : tables) {
84+
table.readLock();
85+
}
86+
try {
87+
for (OlapTable table : tables) {
88+
table.versionWriteLock();
89+
}
90+
try {
91+
for (OlapTable table : tables) {
92+
table.invalidateCachedTableVersion();
93+
for (Partition partition : table.getAllPartitions()) {
94+
((CloudPartition) partition).invalidateCachedVisibleVersion();
95+
}
96+
}
97+
} finally {
98+
for (int i = tables.size() - 1; i >= 0; i--) {
99+
tables.get(i).versionWriteUnlock();
100+
}
101+
}
102+
} finally {
103+
for (int i = tables.size() - 1; i >= 0; i--) {
104+
tables.get(i).readUnlock();
105+
}
106+
}
107+
LOG.info("Invalidated cloud version caches for tables {}",
108+
tables.stream().map(OlapTable::getId).collect(Collectors.toList()));
109+
}
110+
60111
// master FE send sync version rpc to other FEs
61112
public void pushVersionAsync(long dbId, OlapTable table, long version) {
62113
if (!Config.cloud_enable_version_syncer) {
@@ -153,17 +204,25 @@ public void syncVersionAsync(TFrontendSyncCloudVersionRequest request) {
153204
}
154205
// only update table version
155206
if (request.getPartitionVersionInfos().isEmpty()) {
207+
List<OlapTable> invalidatedTables = new ArrayList<>();
156208
request.getTableVersionInfos().forEach(tableVersionInfo -> {
157209
Table table = db.getTableNullable(tableVersionInfo.getTableId());
158210
if (table == null || !table.isManagedTable()) {
159211
return;
160212
}
161213
OlapTable olapTable = (OlapTable) table;
214+
if (tableVersionInfo.getVersion() == INVALIDATE_VERSION_CACHE) {
215+
invalidatedTables.add(olapTable);
216+
return;
217+
}
162218
olapTable.setCachedTableVersion(tableVersionInfo.getVersion());
163219
if (LOG.isDebugEnabled()) {
164220
LOG.debug("Update tableId: {}, version: {}", olapTable.getId(), tableVersionInfo.getVersion());
165221
}
166222
});
223+
if (!invalidatedTables.isEmpty()) {
224+
invalidateVersionCaches(invalidatedTables);
225+
}
167226
return;
168227
}
169228
// update partition and table version

‎fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java‎

Lines changed: 34 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@
4848
import java.util.HashMap;
4949
import java.util.List;
5050
import java.util.Map;
51+
import java.util.concurrent.atomic.AtomicLong;
5152
import java.util.concurrent.locks.ReentrantLock;
5253
import java.util.stream.Collectors;
5354

@@ -66,6 +67,10 @@ public class CloudPartition extends Partition {
6667
// This value is set when get the version from meta-service, 0 means version is not cached yet
6768
private volatile long lastVersionCachedTimeMs = 0;
6869

70+
// Only an MS read started after invalidation may make this cache valid again.
71+
private final AtomicLong versionCacheEpoch = new AtomicLong();
72+
private final AtomicLong refreshedVersionCacheEpoch = new AtomicLong();
73+
6974
private ReentrantLock lock = new ReentrantLock(true);
7075

7176
public CloudPartition(long id, String name, MaterializedIndex baseIndex,
@@ -123,9 +128,13 @@ public long getCachedVisibleVersion() {
123128
return super.getVisibleVersion();
124129
}
125130

131+
public void invalidateCachedVisibleVersion() {
132+
versionCacheEpoch.incrementAndGet();
133+
}
134+
126135
@VisibleForTesting
127136
protected boolean isCachedVersionExpired() {
128-
if (lastVersionCachedTimeMs == 0) {
137+
if (lastVersionCachedTimeMs == 0 || versionCacheEpoch.get() != refreshedVersionCacheEpoch.get()) {
129138
return true;
130139
}
131140
ConnectContext ctx = ConnectContext.get();
@@ -151,6 +160,7 @@ public long getVisibleVersion() {
151160
}
152161

153162
private long getVisibleVersionFromMs(boolean waitForPendingTxns) {
163+
long cacheEpoch = versionCacheEpoch.get();
154164
if (LOG.isDebugEnabled()) {
155165
LOG.debug("getVisibleVersionFromMs use CloudPartition {}, waitForPendingTxns: {}",
156166
super.getName(), waitForPendingTxns);
@@ -182,6 +192,7 @@ private long getVisibleVersionFromMs(boolean waitForPendingTxns) {
182192
LOG.debug("get version from meta service, version: {}, partition: {}", version, super.getId());
183193
}
184194
setCachedVisibleVersion(version, mTime);
195+
refreshedVersionCacheEpoch.accumulateAndGet(cacheEpoch, Math::max);
185196
return version;
186197
} catch (RpcException e) {
187198
throw new RuntimeException("get version from meta service failed");
@@ -242,25 +253,39 @@ public static List<Long> getSnapshotVisibleVersionFromMs(
242253
List<Long> tableIds = new ArrayList<>();
243254
List<Long> partitionIds = new ArrayList<>();
244255
List<Long> versionUpdateTimesMs = new ArrayList<>();
256+
List<Long> cacheEpochs = new ArrayList<>();
245257
for (CloudPartition partition : partitions) {
246258
dbIds.add(partition.getDbId());
247259
tableIds.add(partition.getTableId());
248260
partitionIds.add(partition.getId());
261+
cacheEpochs.add(partition.versionCacheEpoch.get());
249262
}
250263

251264
List<Long> versions = getSnapshotVisibleVersion(
252265
dbIds, tableIds, partitionIds, versionUpdateTimesMs, waitForPendingTxns, maxAttempts);
253266

254267
// Cache visible version, see hasData() for details.
255268
int size = versions.size();
256-
for (int i = 0; i < size; ++i) {
257-
Long version = versions.get(i);
258-
if (version > Partition.PARTITION_INIT_VERSION) {
259-
// For compatibility, the existing partitions may not have mtime
260-
long mTime = versions.size() == versionUpdateTimesMs.size() ? versionUpdateTimesMs.get(i) : 0;
261-
partitions.get(i).setCachedVisibleVersion(versions.get(i), mTime);
262-
} else { // No data has been written to this partition
263-
partitions.get(i).setCachedVisibleVersion(Partition.PARTITION_INIT_VERSION, System.currentTimeMillis());
269+
List<OlapTable> tables = getTables(partitions);
270+
for (OlapTable table : tables) {
271+
table.versionWriteLock();
272+
}
273+
try {
274+
for (int i = 0; i < size; ++i) {
275+
Long version = versions.get(i);
276+
if (version > Partition.PARTITION_INIT_VERSION) {
277+
// For compatibility, the existing partitions may not have mtime
278+
long mTime = versions.size() == versionUpdateTimesMs.size() ? versionUpdateTimesMs.get(i) : 0;
279+
partitions.get(i).setCachedVisibleVersion(versions.get(i), mTime);
280+
} else { // No data has been written to this partition
281+
partitions.get(i).setCachedVisibleVersion(Partition.PARTITION_INIT_VERSION,
282+
System.currentTimeMillis());
283+
}
284+
partitions.get(i).refreshedVersionCacheEpoch.accumulateAndGet(cacheEpochs.get(i), Math::max);
285+
}
286+
} finally {
287+
for (int i = tables.size() - 1; i >= 0; i--) {
288+
tables.get(i).versionWriteUnlock();
264289
}
265290
}
266291

‎fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -478,6 +478,10 @@ private boolean checkTransactionStateBeforeCommit(long dbId, long transactionId)
478478
+ "] is already aborted. abort reason: " + transactionState.getReason());
479479
} else if (transactionState.getTransactionStatus() == TransactionStatus.COMMITTED
480480
|| transactionState.getTransactionStatus() == TransactionStatus.VISIBLE) {
481+
if (transactionState.getTransactionStatus() == TransactionStatus.VISIBLE) {
482+
((CloudEnv) Env.getCurrentEnv()).getCloudFEVersionSynchronizer()
483+
.invalidateVersionCaches(dbId, transactionState.getTableIdList());
484+
}
481485
LOG.info("txn={}, status={} not need to calculate delete bitmap again, just return ",
482486
transactionId,
483487
transactionState.getTransactionStatus().toString());
@@ -582,6 +586,13 @@ private Map<Long, List<Long>> updateVersion(CommitTxnResponse commitTxnResponse)
582586
long txnId = commitTxnResponse.getTxnInfo().getTxnId();
583587
int totalPartitionNum = commitTxnResponse.getPartitionIdsList().size();
584588
if (totalPartitionNum == 0 && commitTxnResponse.getTableStatsList().isEmpty()) {
589+
// A visible retry can contain only txn_info. Invalidate locally without another commit-time RPC;
590+
// subsequent reads fetch fresh versions, including the table version used to validate SQL caches.
591+
if (commitTxnResponse.getTxnInfo().getStatus() == TxnStatusPB.TXN_STATUS_VISIBLE
592+
&& !(commitTxnResponse.getIsLazyCommit() && commitTxnResponse.getIsLazyCommitIncomplete())) {
593+
((CloudEnv) Env.getCurrentEnv()).getCloudFEVersionSynchronizer()
594+
.invalidateVersionCaches(dbId, commitTxnResponse.getTxnInfo().getTableIdsList());
595+
}
585596
return Collections.emptyMap();
586597
}
587598
Env env = Env.getCurrentEnv();

0 commit comments

Comments
 (0)