This is an automated email from the ASF dual-hosted git repository.
hello-stephen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 883bf91da0c [fix](cloud) Avoid caching partition versions with pending
transactions (#68104)
883bf91da0c is described below
commit 883bf91da0c88ce6d9436dd1e0f3d1cc48ed6d00
Author: meiyi <[email protected]>
AuthorDate: Fri Oct 23 15:20:01 2026 +0800
[fix](cloud) Avoid caching partition versions with pending transactions
(#68104)
Non-waiting version reads could cache an older partition version while a
lazy commit was still pending.
Return per-partition pending state alongside scalar and batch versions.
FE keeps pending responses uncached and restores cache validity only for
the epoch captured before the RPC.
---------
Co-authored-by: Gavin Chou <[email protected]>
Co-authored-by: Gavin Chou <[email protected]>
---
cloud/src/meta-service/meta_service.cpp | 14 +-
cloud/test/meta_service_versioned_read_test.cpp | 84 ++++++
.../apache/doris/cloud/catalog/CloudPartition.java | 78 ++++--
.../cloud/catalog/CloudSyncVersionDaemon.java | 26 +-
.../transaction/CloudGlobalTransactionMgr.java | 3 +-
.../doris/common/proc/PartitionsProcDir.java | 2 +-
.../java/org/apache/doris/qe/SessionVariable.java | 7 +-
.../catalog/CloudPartitionPendingVersionTest.java | 303 +++++++++++++++++++++
.../doris/cloud/catalog/CloudPartitionTest.java | 5 +
.../transaction/CloudGlobalTransactionMgrTest.java | 37 ++-
.../org/apache/doris/qe/SessionVariablesTest.java | 18 ++
gensrc/proto/cloud.proto | 4 +
12 files changed, 547 insertions(+), 34 deletions(-)
diff --git a/cloud/src/meta-service/meta_service.cpp
b/cloud/src/meta-service/meta_service.cpp
index 3b691083730..452ccc8252e 100644
--- a/cloud/src/meta-service/meta_service.cpp
+++ b/cloud/src/meta-service/meta_service.cpp
@@ -293,6 +293,7 @@ void
MetaServiceImpl::get_version(::google::protobuf::RpcController* controller,
TxnErrorCode err = reader.get_partition_version(partition_id,
&partition_version,
nullptr, false,
&source_instance);
if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
+ response->add_has_pending_txns(false);
msg = "partition version not found";
code = MetaServiceCode::VERSION_NOT_FOUND;
return;
@@ -311,6 +312,7 @@ void
MetaServiceImpl::get_version(::google::protobuf::RpcController* controller,
}
partition_version = std::move(pending_versions.front());
}
+
response->add_has_pending_txns(!partition_version.pending_txn_ids().empty());
if (!partition_version.has_version()) {
msg = "partition version not found";
code = MetaServiceCode::VERSION_NOT_FOUND;
@@ -369,6 +371,7 @@ void
MetaServiceImpl::get_version(::google::protobuf::RpcController* controller,
}
version_pb = std::move(pending_versions.front());
}
+
response->add_has_pending_txns(!version_pb.pending_txn_ids().empty());
if (!version_pb.has_version()) {
msg = "not found";
code = MetaServiceCode::VERSION_NOT_FOUND;
@@ -382,6 +385,9 @@ void
MetaServiceImpl::get_version(::google::protobuf::RpcController* controller,
TEST_SYNC_POINT_CALLBACK("get_version_code", &code);
return;
} else if (err == TxnErrorCode::TXN_KEY_NOT_FOUND) {
+ if (!is_table_version) {
+ response->add_has_pending_txns(false);
+ }
msg = "not found";
code = MetaServiceCode::VERSION_NOT_FOUND;
return;
@@ -441,6 +447,7 @@ void
MetaServiceImpl::batch_get_version(::google::protobuf::RpcController* contr
response->clear_table_ids();
response->clear_versions();
response->clear_db_ids();
+ response->clear_has_pending_txns();
}
return;
}
@@ -545,7 +552,8 @@ void
MetaServiceImpl::batch_get_version(::google::protobuf::RpcController* contr
}
}
for (auto& version_pb : partition_versions) {
- response->add_versions(version_pb.version());
+ response->add_versions(version_pb.has_version() ?
version_pb.version() : -1);
+
response->add_has_pending_txns(!version_pb.pending_txn_ids().empty());
response->add_version_update_time_ms(version_pb.update_time_ms());
response->add_commit_tsos(version_pb.commit_tso());
}
@@ -558,6 +566,7 @@ void
MetaServiceImpl::batch_get_version(::google::protobuf::RpcController* contr
response->clear_versions();
response->clear_version_update_time_ms();
response->clear_commit_tsos();
+ response->clear_has_pending_txns();
}
}
@@ -707,7 +716,8 @@ std::pair<MetaServiceCode, std::string>
MetaServiceImpl::batch_get_partition_ver
}
}
for (auto& version_pb : partition_versions) {
- response->add_versions(version_pb.version());
+ response->add_versions(version_pb.has_version() ?
version_pb.version() : -1);
+
response->add_has_pending_txns(!version_pb.pending_txn_ids().empty());
response->add_version_update_time_ms(version_pb.update_time_ms());
response->add_commit_tsos(version_pb.commit_tso());
}
diff --git a/cloud/test/meta_service_versioned_read_test.cpp
b/cloud/test/meta_service_versioned_read_test.cpp
index 7ba77a1d3aa..3fc8f833a91 100644
--- a/cloud/test/meta_service_versioned_read_test.cpp
+++ b/cloud/test/meta_service_versioned_read_test.cpp
@@ -591,6 +591,90 @@ TEST(MetaServiceVersionedReadTest, GetVersion) {
}
}
+// Exercise both storage layouts and both scalar/batch wire responses.
+TEST(MetaServiceVersionedReadTest, GetVersionPendingState) {
+ for (bool versioned_read : {false, true}) {
+ SCOPED_TRACE(versioned_read);
+ auto service = get_meta_service(false);
+ std::string instance_id = "pending_version_instance";
+ MOCK_GET_INSTANCE_ID(instance_id);
+ if (versioned_read) {
+ create_and_refresh_instance(service.get(), instance_id);
+ }
+ std::unique_ptr<Transaction> txn;
+ ASSERT_EQ(service->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
+ for (int64_t partition_id : {1, 2, 3}) {
+ VersionPB version;
+ if (partition_id != 3) {
+ version.set_version(partition_id == 1 ? 12 : 7);
+ }
+ if (partition_id != 2) {
+ version.add_pending_txn_ids(100);
+ }
+ version.set_update_time_ms(2000);
+ version.set_commit_tso(3000);
+ std::string value = version.SerializeAsString();
+ txn->put(partition_version_key({instance_id, 1, 1, partition_id}),
value);
+ versioned_put(txn.get(),
versioned::partition_version_key({instance_id, partition_id}),
+ value);
+ }
+ // Model waiting on an observed pending version after its transaction
has become visible.
+ TxnIndexPB index;
+ index.mutable_tablet_index()->set_db_id(1);
+ txn->put(txn_index_key({instance_id, 100}), index.SerializeAsString());
+ TxnInfoPB info;
+ info.set_txn_id(100);
+ info.set_db_id(1);
+ info.set_status(TxnStatusPB::TXN_STATUS_VISIBLE);
+ txn->put(txn_info_key({instance_id, 1, 100}),
info.SerializeAsString());
+ ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
+
+ for (bool wait : {false, true}) {
+ SCOPED_TRACE(wait);
+ // Mix pending, non-pending, first-write pending, and absent
partitions in request order.
+ std::vector<int64_t> ids {2, 1, 3, 4};
+ std::vector<int64_t> expected_versions {7, wait ? 13 : 12, wait ?
2 : -1, -1};
+ std::vector<bool> expected_pending {false, !wait, !wait, false};
+ GetVersionRequest batch;
+ batch.set_cloud_unique_id("test_cloud_unique_id");
+ batch.set_batch_mode(true);
+ batch.set_wait_for_pending_txn(wait);
+ for (size_t i = 0; i < ids.size(); ++i) {
+ GetVersionRequest req;
+ req.set_cloud_unique_id("test_cloud_unique_id");
+ req.set_db_id(1);
+ req.set_table_id(1);
+ req.set_partition_id(ids[i]);
+ req.set_wait_for_pending_txn(wait);
+ brpc::Controller ctrl;
+ GetVersionResponse resp;
+ service->get_version(&ctrl, &req, &resp, nullptr);
+ ASSERT_EQ(resp.status().code(), expected_versions[i] == -1
+ ?
MetaServiceCode::VERSION_NOT_FOUND
+ : MetaServiceCode::OK);
+ ASSERT_EQ(resp.has_pending_txns_size(), 1);
+ EXPECT_EQ(resp.has_pending_txns(0), expected_pending[i]);
+ if (expected_versions[i] != -1) {
+ EXPECT_EQ(resp.version(), expected_versions[i]);
+ }
+ batch.add_db_ids(1);
+ batch.add_table_ids(1);
+ batch.add_partition_ids(ids[i]);
+ }
+ brpc::Controller ctrl;
+ GetVersionResponse resp;
+ service->get_version(&ctrl, &batch, &resp, nullptr);
+ ASSERT_EQ(resp.status().code(), MetaServiceCode::OK) <<
resp.DebugString();
+ ASSERT_EQ(resp.versions_size(), ids.size());
+ ASSERT_EQ(resp.has_pending_txns_size(), ids.size());
+ for (size_t i = 0; i < ids.size(); ++i) {
+ EXPECT_EQ(resp.versions(i), expected_versions[i]);
+ EXPECT_EQ(resp.has_pending_txns(i), expected_pending[i]);
+ }
+ }
+ }
+}
+
TEST(MetaServiceVersionedReadTest, BatchGetVersion) {
struct TestCase {
std::vector<int64_t> table_ids;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java
index 9c788a9ab7f..5bbc3a89646 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java
@@ -160,10 +160,16 @@ public class CloudPartition extends Partition {
return getCachedVisibleVersion();
}
- return getVisibleVersionFromMs(false);
+ return getVisibleVersionFromMs(shouldWaitForPendingTxns());
}
- private long getVisibleVersionFromMs(boolean waitForPendingTxns) {
+ private static boolean shouldWaitForPendingTxns() {
+ ConnectContext ctx = ConnectContext.get();
+ return ctx == null ?
VariableMgr.getDefaultSessionVariable().cloudGetVersionWaitForPendingTxn
+ : ctx.getSessionVariable().cloudGetVersionWaitForPendingTxn;
+ }
+
+ public long getVisibleVersionFromMs(boolean waitForPendingTxns) {
long cacheEpoch = versionCacheEpoch.get();
if (LOG.isDebugEnabled()) {
LOG.debug("getVisibleVersionFromMs use CloudPartition {},
waitForPendingTxns: {}",
@@ -192,8 +198,7 @@ public class CloudPartition extends Partition {
LOG.debug("get version from meta service, version: {},
tso: {}, partition: {}",
version, tso, super.getId());
}
- setCachedVisibleVersion(version, mTime, tso);
- refreshedVersionCacheEpoch.accumulateAndGet(cacheEpoch,
Math::max);
+ cacheVisibleVersion(version, mTime, tso, cacheEpoch,
isVersionCacheable(resp, 0, 1));
return version;
} else {
assert resp.getStatus().getCode() ==
MetaServiceCode.VERSION_NOT_FOUND;
@@ -203,14 +208,37 @@ public class CloudPartition extends Partition {
if (LOG.isDebugEnabled()) {
LOG.debug("get version from meta service, version: {},
partition: {}", version, super.getId());
}
- setCachedVisibleVersion(version, mTime);
- refreshedVersionCacheEpoch.accumulateAndGet(cacheEpoch, Math::max);
+ cacheVisibleVersion(version, mTime, -1, cacheEpoch,
isVersionCacheable(resp, 0, 1));
return version;
} catch (RpcException e) {
throw new RuntimeException("get version from meta service failed");
}
}
+ private void cacheVisibleVersion(long version, long updateTimeMs, long
tso, long cacheEpoch, boolean cacheable) {
+ if (!cacheable) {
+ // Invalidate before touching the timestamp, even when the cache
was valid before this read.
+ invalidateCachedVisibleVersion();
+ }
+ setCachedVisibleVersion(version, updateTimeMs, tso);
+ if (cacheable) {
+ // A response must not acknowledge an invalidation that happened
after its request started.
+ refreshedVersionCacheEpoch.accumulateAndGet(cacheEpoch, Math::max);
+ }
+ }
+
+ private static boolean isVersionCacheable(Cloud.GetVersionResponse
response, int index, int size) {
+ if (response.getHasPendingTxnsCount() == 0) {
+ // Older MS does not report pending state. Preserve its existing
cache behavior.
+ if (index == 0 && LOG.isDebugEnabled()) {
+ LOG.debug("MS version response has no pending transaction
state; using legacy cache behavior"
+ + " for {} partitions", size);
+ }
+ return true;
+ }
+ return response.getHasPendingTxnsCount() == size &&
!response.getHasPendingTxns(index);
+ }
+
// Select the non-empty partitions and return the ids.
public static List<Long> selectNonEmptyPartitionIds(List<CloudPartition>
partitions) {
return selectNonEmptyPartitionIds(partitions, false);
@@ -261,6 +289,10 @@ public class CloudPartition extends Partition {
// Get visible version from the specified partitions;
//
// Return the visible version in order of the specified partition ids
+ public static List<Long>
getSnapshotVisibleVersionFromMs(List<CloudPartition> partitions) throws
RpcException {
+ return getSnapshotVisibleVersionFromMs(partitions,
shouldWaitForPendingTxns());
+ }
+
public static List<Long> getSnapshotVisibleVersionFromMs(
List<CloudPartition> partitions, boolean waitForPendingTxns)
throws RpcException {
return getSnapshotVisibleVersionFromMs(
@@ -278,6 +310,7 @@ public class CloudPartition extends Partition {
List<Long> partitionIds = new ArrayList<>();
List<Long> versionUpdateTimesMs = new ArrayList<>();
List<Long> commitTsos = new ArrayList<>();
+ List<Boolean> cacheableVersions = new ArrayList<>();
List<Long> cacheEpochs = new ArrayList<>();
for (CloudPartition partition : partitions) {
dbIds.add(partition.getDbId());
@@ -287,7 +320,8 @@ public class CloudPartition extends Partition {
}
List<Long> versions = getSnapshotVisibleVersion(
- dbIds, tableIds, partitionIds, versionUpdateTimesMs,
commitTsos, waitForPendingTxns, maxAttempts);
+ dbIds, tableIds, partitionIds, versionUpdateTimesMs,
commitTsos, cacheableVersions,
+ waitForPendingTxns, maxAttempts);
// Cache visible version, see hasData() for details.
int size = versions.size();
@@ -303,12 +337,12 @@ public class CloudPartition extends Partition {
// For compatibility, the existing partitions may not have
mtime
long mTime = versions.size() ==
versionUpdateTimesMs.size() ? versionUpdateTimesMs.get(i) : 0;
long tso = hasCommitTsos ? commitTsos.get(i) : -1;
- partitions.get(i).setCachedVisibleVersion(versions.get(i),
mTime, tso);
+ partitions.get(i).cacheVisibleVersion(version, mTime, tso,
+ cacheEpochs.get(i), cacheableVersions.get(i));
} else { // No data has been written to this partition
-
partitions.get(i).setCachedVisibleVersion(Partition.PARTITION_INIT_VERSION,
- System.currentTimeMillis());
+
partitions.get(i).cacheVisibleVersion(Partition.PARTITION_INIT_VERSION,
+ System.currentTimeMillis(), -1,
cacheEpochs.get(i), cacheableVersions.get(i));
}
-
partitions.get(i).refreshedVersionCacheEpoch.accumulateAndGet(cacheEpochs.get(i),
Math::max);
}
} finally {
for (int i = tables.size() - 1; i >= 0; i--) {
@@ -353,7 +387,7 @@ public class CloudPartition extends Partition {
?
VariableMgr.getDefaultSessionVariable().cloudPartitionVersionCacheTtlMs
: ctx.getSessionVariable().cloudPartitionVersionCacheTtlMs;
if (cloudPartitionVersionCacheTtlMs <= 0) { // No cached versions will
be used
- return getSnapshotVisibleVersionFromMs(partitions, false);
+ return getSnapshotVisibleVersionFromMs(partitions);
}
// partitionId -> cachedVersion
@@ -384,8 +418,7 @@ public class CloudPartition extends Partition {
List<Long> versions = null;
if (!expiredPartitions.isEmpty()) { // Not all partition versions are
from cache
- versions = getSnapshotVisibleVersionFromMs(
- expiredPartitions, /*waitForPendingTxns=*/false); // Get
the rest versions from meta-service
+ versions = getSnapshotVisibleVersionFromMs(expiredPartitions); //
Get the rest versions from meta-service
}
int verMsIdx = 0;
for (Pair<Long, Long> v : allVersions) { // ATTN: keep the assigning
order!!!
@@ -405,7 +438,8 @@ public class CloudPartition extends Partition {
//
// Return the visible version in order of the specified partition ids
private static List<Long> getSnapshotVisibleVersion(List<Long> dbIds,
List<Long> tableIds, List<Long> partitionIds,
- List<Long> versionUpdateTimesMs, List<Long> commitTsos, boolean
waitForPendingTxns, int maxAttempts)
+ List<Long> versionUpdateTimesMs, List<Long> commitTsos,
List<Boolean> cacheableVersions,
+ boolean waitForPendingTxns, int maxAttempts)
throws RpcException {
assert dbIds.size() == partitionIds.size() :
"partition ids size: " + partitionIds.size() + " should equals
to db ids size: " + dbIds.size();
@@ -449,11 +483,15 @@ public class CloudPartition extends Partition {
commitTsos.addAll(resp.getCommitTsosList());
}
- ArrayList<Long> news = new ArrayList<>();
- for (Long v : versions) { // -1 means version NOT FOUND ==> no data
has been written
- news.add(v == -1 ? Partition.PARTITION_INIT_VERSION : v);
- }
- return news;
+ int size = versions.size();
+ List<Long> normalizedVersions = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ cacheableVersions.add(isVersionCacheable(resp, i, size));
+ long version = versions.get(i);
+ // -1 means version NOT FOUND.
+ normalizedVersions.add(version == -1 ?
Partition.PARTITION_INIT_VERSION : version);
+ }
+ return normalizedVersions;
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java
index 6d485d959bc..4b9ab39d8bf 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java
@@ -135,13 +135,7 @@ public class CloudSyncVersionDaemon extends MasterDaemon {
OlapTable table = tables.get(i);
long version = versions.get(i);
if (version > table.getCachedTableVersion()) {
- tableVersionMap.compute(table, (k, v) -> {
- if (v == null || version > v) {
- return version;
- } else {
- return v;
- }
- });
+ prepareTableForVersionSync(tableVersionMap, table,
version);
} else {
// update lastTableVersionCachedTimeMs
table.setCachedTableVersion(version);
@@ -154,6 +148,24 @@ public class CloudSyncVersionDaemon extends MasterDaemon {
});
}
+ private void prepareTableForVersionSync(Map<OlapTable, Long>
tableVersionMap, OlapTable table, long version) {
+ // Stop serving potentially stale partition caches while refresh is
queued or fails.
+ table.readLock();
+ try {
+ table.versionWriteLock();
+ try {
+ for (Partition partition : table.getAllPartitions()) {
+ ((CloudPartition)
partition).invalidateCachedVisibleVersion();
+ }
+ } finally {
+ table.versionWriteUnlock();
+ }
+ } finally {
+ table.readUnlock();
+ }
+ tableVersionMap.merge(table, version, Math::max);
+ }
+
private void syncPartitionVersion(Map<OlapTable, Long> tableVersionMap) {
Set<Long> failedTables = ConcurrentHashMap.newKeySet();
List<Future<Void>> futures = new ArrayList<>();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
index 7716e51599d..ba5afbc2f6c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java
@@ -1192,7 +1192,8 @@ public class CloudGlobalTransactionMgr implements
GlobalTransactionMgrIface {
if (!Config.calc_delete_bitmap_get_versions_in_batch) {
Map<Long, Long> partitionToVersions = Maps.newHashMap();
partitionMap.forEach((key, value) -> {
- long visibleVersion = value.getVisibleVersion();
+ long visibleVersion = ((CloudPartition)
value).getVisibleVersionFromMs(
+
Config.calc_delete_bitmap_get_versions_waiting_for_pending_txns);
long newVersion = visibleVersion <= 0 ? 2 : visibleVersion + 1;
partitionToVersions.put(key, newVersion);
});
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/proc/PartitionsProcDir.java
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/PartitionsProcDir.java
index e0a179cdb8f..83cbd31e9e1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/common/proc/PartitionsProcDir.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/common/proc/PartitionsProcDir.java
@@ -422,7 +422,7 @@ public class PartitionsProcDir implements ProcDirInterface {
for (int start = 0; start < partitions.size(); start +=
batchSize) {
int end = Math.min(start + batchSize, partitions.size());
List<CloudPartition> batch = partitions.subList(start,
end);
-
partitionVersions.addAll(CloudPartition.getSnapshotVisibleVersionFromMs(batch,
false));
+
partitionVersions.addAll(CloudPartition.getSnapshotVisibleVersionFromMs(batch,
true));
}
} catch (RpcException e) {
LOG.warn("get partition versions failed for table: {}",
olapTable, e);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 542ddbc5174..6c65b9ddaf4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -880,6 +880,7 @@ public class SessionVariable implements Serializable,
Writable {
public static final String COMPUTE_GROUP = "compute_group";
public static final String DISABLE_EMPTY_PARTITION_PRUNE =
"disable_empty_partition_prune";
public static final String CLOUD_FORCE_SYNC_VERSION =
"cloud_force_sync_version";
+ public static final String CLOUD_GET_VERSION_WAIT_FOR_PENDING_TXN =
"cloud_get_version_wait_for_pending_txn";
public static final String CLOUD_PARTITION_VERSION_CACHE_TTL_MS =
"cloud_partition_version_cache_ttl_ms";
public static final String CLOUD_TABLE_VERSION_CACHE_TTL_MS =
@@ -3301,11 +3302,15 @@ public class SessionVariable implements Serializable,
Writable {
@VarAttrDef.VarAttr(name = CLOUD_FORCE_SYNC_TABLET_STATS, needForward =
true)
public boolean cloudForceSyncTabletStats = false;
@VarAttrDef.VarAttr(name = CLOUD_PARTITION_VERSION_CACHE_TTL_MS)
- public long cloudPartitionVersionCacheTtlMs = Long.MAX_VALUE;
+ public long cloudPartitionVersionCacheTtlMs = 10 * 60 * 1000L;
@VarAttrDef.VarAttr(name = CLOUD_TABLE_VERSION_CACHE_TTL_MS)
public long cloudTableVersionCacheTtlMs = Long.MAX_VALUE;
@VarAttrDef.VarAttr(name = CLOUD_FORCE_SYNC_VERSION, needForward = true)
public boolean cloudForceSyncVersion = false;
+ @VarAttrDef.VarAttr(name = CLOUD_GET_VERSION_WAIT_FOR_PENDING_TXN,
needForward = true,
+ description = "Whether Cloud partition version RPCs wait for
pending transactions. "
+ + "Does not bypass the version cache or override
explicitly specified waiting policies.")
+ public boolean cloudGetVersionWaitForPendingTxn = false;
// CLOUD_VARIABLES_END
// fetch remote schema rpc timeout
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionPendingVersionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionPendingVersionTest.java
new file mode 100644
index 00000000000..a6af0c5cd25
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionPendingVersionTest.java
@@ -0,0 +1,303 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.cloud.catalog;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.cloud.proto.Cloud;
+import org.apache.doris.cloud.proto.Cloud.MetaServiceCode;
+import org.apache.doris.cloud.rpc.VersionHelper;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.qe.VariableMgr;
+import org.apache.doris.rpc.RpcException;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class CloudPartitionPendingVersionTest {
+ private ConnectContext previousContext;
+ private MockedStatic<Env> env;
+ private MockedStatic<VersionHelper> versions;
+ private CloudPartition partition;
+ private Cloud.GetVersionResponse response;
+ private Cloud.GetVersionRequest lastRequest;
+ private Runnable beforeResponse;
+ private int requests;
+
+ @BeforeEach
+ public void setUp() {
+ previousContext = ConnectContext.get();
+ ConnectContext ctx = new ConnectContext();
+ ctx.setSessionVariable(new SessionVariable());
+ ctx.getSessionVariable().cloudPartitionVersionCacheTtlMs =
Long.MAX_VALUE;
+ ctx.setThreadLocalInfo();
+ env = Mockito.mockStatic(Env.class);
+
env.when(Env::getCurrentInternalCatalog).thenReturn(Mockito.mock(InternalCatalog.class));
+ partition = CloudPartitionTest.createPartition(1, 2, 3);
+ partition.setCachedVisibleVersion(12, 1000);
+ versions = Mockito.mockStatic(VersionHelper.class);
+ versions.when(() ->
VersionHelper.getVersionFromMeta(Mockito.any(Cloud.GetVersionRequest.class)))
+ .thenAnswer(invocation ->
nextResponse(invocation.getArgument(0)));
+ versions.when(() -> VersionHelper.getVersionFromMeta(
+ Mockito.any(Cloud.GetVersionRequest.class), Mockito.anyInt()))
+ .thenAnswer(invocation ->
nextResponse(invocation.getArgument(0)));
+ }
+
+ @AfterEach
+ public void tearDown() {
+ versions.close();
+ env.close();
+ if (previousContext == null) {
+ ConnectContext.remove();
+ } else {
+ previousContext.setThreadLocalInfo();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testPendingReadCannotRestoreInvalidatedCache(boolean batch)
throws Exception {
+ partition.invalidateCachedVisibleVersion();
+ response = response(12).addHasPendingTxns(true).build();
+ Assertions.assertEquals(12, read(batch));
+ Assertions.assertTrue(partition.isCachedVersionExpired());
+ // A later ordinary commit must not hide the pending update by
refreshing the cache timestamp.
+ partition.setCachedVisibleVersion(12, 2000);
+ Assertions.assertEquals(12, read(batch));
+ Assertions.assertEquals(2, requests);
+ Assertions.assertTrue(partition.isCachedVersionExpired());
+
+ response = response(13).addHasPendingTxns(false).build();
+ Assertions.assertEquals(13, read(batch));
+ Assertions.assertFalse(partition.isCachedVersionExpired());
+ Assertions.assertEquals(13, read(batch));
+ Assertions.assertEquals(3, requests, "A non-pending query response may
restore caching without a daemon");
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testPendingReadInvalidatesPreviouslyValidCache(boolean batch)
throws Exception {
+ Assertions.assertFalse(partition.isCachedVersionExpired());
+
ConnectContext.get().getSessionVariable().cloudPartitionVersionCacheTtlMs = 0;
+ response = response(12).addHasPendingTxns(true).build();
+ Assertions.assertEquals(12, read(batch));
+
ConnectContext.get().getSessionVariable().cloudPartitionVersionCacheTtlMs =
Long.MAX_VALUE;
+ Assertions.assertTrue(partition.isCachedVersionExpired());
+ }
+
+ @Test
+ public void testBatchInvalidatesOnlyPendingPartitions() throws Exception {
+ CloudPartition other = CloudPartitionTest.createPartition(4, 2, 3);
+ other.setCachedVisibleVersion(7, 1000);
+ response =
response(12).addVersions(8).addHasPendingTxns(true).addHasPendingTxns(false).build();
+ // Forced MS reads may observe pending while the previous cache was
still valid.
+ Assertions.assertEquals(List.of(12L, 8L),
+
CloudPartition.getSnapshotVisibleVersionFromMs(List.of(partition, other),
false));
+ Assertions.assertTrue(partition.isCachedVersionExpired());
+ Assertions.assertFalse(other.isCachedVersionExpired());
+ response = response(13).addHasPendingTxns(false).build();
+ Assertions.assertEquals(List.of(13L, 8L),
+ CloudPartition.getSnapshotVisibleVersion(List.of(partition,
other)));
+ Assertions.assertFalse(partition.isCachedVersionExpired());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testFirstPendingWriteDoesNotCacheEmptyPartition(boolean batch)
throws Exception {
+ partition = CloudPartitionTest.createPartition(1, 2, 3);
+ response =
response(-1).setStatus(Cloud.MetaServiceResponseStatus.newBuilder()
+ .setCode(batch ? MetaServiceCode.OK :
MetaServiceCode.VERSION_NOT_FOUND))
+ .addHasPendingTxns(true).build();
+ Assertions.assertEquals(1, read(batch));
+ Assertions.assertTrue(partition.isCachedVersionExpired());
+ response = response(2).addHasPendingTxns(false).build();
+ Assertions.assertEquals(2, read(batch));
+ Assertions.assertFalse(partition.isCachedVersionExpired());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testNonPendingResponsePreservesConcurrentInvalidation(boolean
batch) throws Exception {
+ partition.invalidateCachedVisibleVersion();
+ response = response(13).addHasPendingTxns(false).build();
+ beforeResponse = partition::invalidateCachedVisibleVersion;
+ Assertions.assertEquals(13, read(batch));
+ Assertions.assertTrue(partition.isCachedVersionExpired());
+ beforeResponse = null;
+ Assertions.assertEquals(13, read(batch));
+ Assertions.assertFalse(partition.isCachedVersionExpired());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testOlderMsPreservesLegacyCacheBehavior(boolean batch) throws
Exception {
+ partition.invalidateCachedVisibleVersion();
+ response = response(13).build();
+ Assertions.assertEquals(13, read(batch));
+ Assertions.assertFalse(partition.isCachedVersionExpired());
+ Assertions.assertEquals(13, read(batch));
+ Assertions.assertEquals(1, requests, "An older MS response must still
allow cache hits without waiting");
+
+ partition.invalidateCachedVisibleVersion();
+ beforeResponse = partition::invalidateCachedVisibleVersion;
+ Assertions.assertEquals(13, read(batch));
+ Assertions.assertTrue(partition.isCachedVersionExpired(),
+ "Legacy compatibility must not acknowledge invalidation after
the RPC started");
+ }
+
+ @Test
+ public void testExplicitPendingOverridesWaitFlag() throws Exception {
+ response = response(12).addHasPendingTxns(true).build();
+ CloudPartition.getSnapshotVisibleVersionFromMs(List.of(partition),
true);
+ Assertions.assertTrue(partition.isCachedVersionExpired());
+ response = response(13).addHasPendingTxns(false).build();
+ CloudPartition.getSnapshotVisibleVersionFromMs(List.of(partition),
true);
+ Assertions.assertFalse(partition.isCachedVersionExpired());
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false,false,false", "false,true,false", "true,false,false",
"true,true,false",
+ "false,false,true", "false,true,true", "true,false,true",
"true,true,true"})
+ public void testSessionWaitFlagPreservesCachePolicy(
+ boolean batch, boolean waitForPendingTxns, boolean cacheEnabled)
throws Exception {
+ SessionVariable session = ConnectContext.get().getSessionVariable();
+ session.cloudGetVersionWaitForPendingTxn = waitForPendingTxns;
+ session.cloudPartitionVersionCacheTtlMs = cacheEnabled ?
Long.MAX_VALUE : 0;
+ response = response(13).addHasPendingTxns(false).build();
+ if (cacheEnabled) {
+ Assertions.assertEquals(12, read(batch));
+ Assertions.assertEquals(0, requests, "Enabling waiting must not
bypass a valid cache");
+ partition.invalidateCachedVisibleVersion();
+ }
+ Assertions.assertEquals(13, read(batch));
+ Assertions.assertEquals(1, requests);
+ Assertions.assertEquals(batch, lastRequest.getBatchMode());
+ Assertions.assertEquals(waitForPendingTxns,
lastRequest.getWaitForPendingTxn());
+ Assertions.assertEquals(13, read(batch));
+ Assertions.assertEquals(cacheEnabled ? 1 : 2, requests);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testForcedReadUsesSessionWaitFlag(boolean waitForPendingTxns)
throws Exception {
+
ConnectContext.get().getSessionVariable().cloudGetVersionWaitForPendingTxn =
waitForPendingTxns;
+ response = response(13).addHasPendingTxns(false).build();
+ Assertions.assertEquals(List.of(13L),
CloudPartition.getSnapshotVisibleVersionFromMs(List.of(partition)));
+ Assertions.assertEquals(1, requests);
+ Assertions.assertEquals(waitForPendingTxns,
lastRequest.getWaitForPendingTxn());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testExplicitWaitFlagOverridesSession(boolean
waitForPendingTxns) throws Exception {
+
ConnectContext.get().getSessionVariable().cloudGetVersionWaitForPendingTxn =
!waitForPendingTxns;
+ response = response(13).addHasPendingTxns(false).build();
+ Assertions.assertEquals(13,
partition.getVisibleVersionFromMs(waitForPendingTxns));
+ Assertions.assertEquals(waitForPendingTxns,
lastRequest.getWaitForPendingTxn());
+ Assertions.assertEquals(List.of(13L),
+
CloudPartition.getSnapshotVisibleVersionFromMs(List.of(partition),
waitForPendingTxns));
+ Assertions.assertEquals(waitForPendingTxns,
lastRequest.getWaitForPendingTxn());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testReadWithoutSessionUsesGlobalWaitFlag(boolean
waitForPendingTxns) throws Exception {
+ SessionVariable defaults = VariableMgr.getDefaultSessionVariable();
+ boolean previousWait = defaults.cloudGetVersionWaitForPendingTxn;
+ try {
+ defaults.cloudGetVersionWaitForPendingTxn = waitForPendingTxns;
+
ConnectContext.get().getSessionVariable().cloudGetVersionWaitForPendingTxn =
!waitForPendingTxns;
+ response = response(13).addHasPendingTxns(false).build();
+ CloudPartition.getSnapshotVisibleVersionFromMs(List.of(partition));
+ Assertions.assertEquals(!waitForPendingTxns,
lastRequest.getWaitForPendingTxn());
+
+ ConnectContext.remove();
+ Assertions.assertEquals(List.of(13L),
CloudPartition.getSnapshotVisibleVersionFromMs(List.of(partition)));
+ Assertions.assertEquals(waitForPendingTxns,
lastRequest.getWaitForPendingTxn());
+ partition.invalidateCachedVisibleVersion();
+ Assertions.assertEquals(13, partition.getVisibleVersion());
+ Assertions.assertEquals(waitForPendingTxns,
lastRequest.getWaitForPendingTxn());
+ } finally {
+ defaults.cloudGetVersionWaitForPendingTxn = previousWait;
+ }
+ }
+
+ @Test
+ public void testDaemonPreparesTableForVersionSync() throws Exception {
+ CloudPartition other = CloudPartitionTest.createPartition(4, 2, 3);
+ other.setCachedVisibleVersion(20, 1000);
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Mockito.when(table.getAllPartitions()).thenReturn(List.of(partition,
other));
+ Map<OlapTable, Long> tablesToSync = new HashMap<>();
+ Method prepare = CloudSyncVersionDaemon.class.getDeclaredMethod(
+ "prepareTableForVersionSync", Map.class, OlapTable.class,
long.class);
+ prepare.setAccessible(true);
+ prepare.invoke(new CloudSyncVersionDaemon(), tablesToSync, table, 3L);
+
+ Assertions.assertTrue(partition.isCachedVersionExpired());
+ Assertions.assertTrue(other.isCachedVersionExpired());
+ Assertions.assertEquals(Map.of(table, 3L), tablesToSync);
+ Assertions.assertEquals(12, partition.getCachedVisibleVersion());
+ Assertions.assertEquals(20, other.getCachedVisibleVersion());
+ versions.verifyNoInteractions();
+ Mockito.verify(table,
Mockito.never()).setCachedTableVersion(Mockito.anyLong());
+ versions.when(() ->
VersionHelper.getVersionFromMeta(Mockito.any(Cloud.GetVersionRequest.class)))
+ .thenThrow(new RpcException("get version", "injected refresh
failure"));
+ Assertions.assertThrows(RuntimeException.class, () ->
partition.getVisibleVersion());
+ Assertions.assertTrue(partition.isCachedVersionExpired(), "A failed
refresh must leave the cache invalid");
+ Assertions.assertTrue(other.isCachedVersionExpired());
+ versions.when(() ->
VersionHelper.getVersionFromMeta(Mockito.any(Cloud.GetVersionRequest.class)))
+ .thenReturn(response(13).addHasPendingTxns(false).build());
+ Assertions.assertEquals(13, partition.getVisibleVersion());
+ Assertions.assertFalse(partition.isCachedVersionExpired());
+ }
+
+ private long read(boolean batch) throws RpcException {
+ return batch ?
CloudPartition.getSnapshotVisibleVersion(List.of(partition)).get(0)
+ : partition.getVisibleVersion();
+ }
+
+ private Cloud.GetVersionResponse nextResponse(Cloud.GetVersionRequest
request) {
+ lastRequest = request;
+ requests++;
+ if (beforeResponse != null) {
+ beforeResponse.run();
+ }
+ return response;
+ }
+
+ private Cloud.GetVersionResponse.Builder response(long version) {
+ return Cloud.GetVersionResponse.newBuilder()
+
.setStatus(Cloud.MetaServiceResponseStatus.newBuilder().setCode(MetaServiceCode.OK))
+
.setVersion(version).addVersions(version).addVersionUpdateTimeMs(2000).addCommitTsos(3000);
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionTest.java
index 9a33dcf70af..63889a798b9 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudPartitionTest.java
@@ -127,6 +127,11 @@ public class CloudPartitionTest {
Cloud.GetVersionResponse.Builder builder =
Cloud.GetVersionResponse.newBuilder();
builder.setVersion(singleVersions.get(callCount[0]));
builder.addAllVersions(batchVersions.get(callCount[0]));
+ Cloud.GetVersionRequest request = invocation.getArgument(0);
+ int size = request.getBatchMode() ? builder.getVersionsCount()
: 1;
+ for (int i = 0; i < size; i++) {
+ builder.addHasPendingTxns(false);
+ }
++callCount[0];
return builder.build();
};
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java
index 8f617c8f219..e7902076eea 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgrTest.java
@@ -88,6 +88,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.function.Executable;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.AdditionalAnswers;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
@@ -1109,7 +1110,8 @@ public class CloudGlobalTransactionMgrTest {
AtomicReference<CompletableFuture<Void>> snapshotReader = new
AtomicReference<>();
try (MockedStatic<VersionHelper> versions = mockVersionHelper()) {
versions.when(() ->
VersionHelper.getVersionFromMeta(Mockito.any())).thenReturn(
-
partitionVersion(4).toBuilder().addVersions(4).addVersionUpdateTimeMs(40).addCommitTsos(400).build());
+
partitionVersion(4).toBuilder().addVersions(4).addVersionUpdateTimeMs(40).addCommitTsos(400)
+ .addHasPendingTxns(false).build());
Mockito.doAnswer(invocation -> {
CountDownLatch started = new CountDownLatch(1);
CompletableFuture<Void> reader = CompletableFuture.runAsync(()
-> {
@@ -1213,6 +1215,37 @@ public class CloudGlobalTransactionMgrTest {
}
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testNonBatchPartitionVersionsBypassCache(boolean
waitForPendingTxns) throws Exception {
+ boolean previousBatch =
Config.calc_delete_bitmap_get_versions_in_batch;
+ boolean previousWait =
Config.calc_delete_bitmap_get_versions_waiting_for_pending_txns;
+ useVersionCaches();
+ CloudPartition partition = addCloudPartition(1000);
+ try (MockedStatic<VersionHelper> versions = mockVersionHelper()) {
+ Config.calc_delete_bitmap_get_versions_in_batch = false;
+ Config.calc_delete_bitmap_get_versions_waiting_for_pending_txns =
waitForPendingTxns;
+ Assertions.assertEquals(2, partition.getVisibleVersion());
+ versions.verifyNoInteractions();
+ versions.when(() ->
VersionHelper.getVersionFromMeta(Mockito.any())).thenAnswer(invocation -> {
+ Cloud.GetVersionRequest request = invocation.getArgument(0);
+ Assertions.assertFalse(request.getBatchMode());
+ Assertions.assertEquals(partition.getId(),
request.getPartitionId());
+ Assertions.assertEquals(waitForPendingTxns,
request.getWaitForPendingTxn());
+ return partitionVersion(4);
+ });
+ Method getVersions =
CloudGlobalTransactionMgr.class.getDeclaredMethod("getPartitionVersions",
Map.class);
+ getVersions.setAccessible(true);
+ Assertions.assertEquals(Map.of(partition.getId(), 5L),
+ getVersions.invoke(masterTransMgr,
Map.of(partition.getId(), partition)));
+ Assertions.assertEquals(4, partition.getCachedVisibleVersion());
+ versions.verify(() ->
VersionHelper.getVersionFromMeta(Mockito.any()));
+ } finally {
+ Config.calc_delete_bitmap_get_versions_in_batch = previousBatch;
+ Config.calc_delete_bitmap_get_versions_waiting_for_pending_txns =
previousWait;
+ }
+ }
+
private CloudPartition addCloudPartition(long tableId) throws Exception {
OlapTable table = new OlapTable(tableId, "version_cache_" + tableId,
List.of(), KeysType.DUP_KEYS,
new PartitionInfo(), new RandomDistributionInfo(1));
@@ -1242,7 +1275,7 @@ public class CloudGlobalTransactionMgrTest {
private Cloud.GetVersionResponse partitionVersion(long version) {
return
Cloud.GetVersionResponse.newBuilder().setVersion(version).addVersions(version)
- .addVersionUpdateTimeMs(version * 10).addCommitTsos(version *
100).build();
+ .addVersionUpdateTimeMs(version * 10).addCommitTsos(version *
100).addHasPendingTxns(false).build();
}
private OlapTable getCloudTable(CloudPartition partition) throws Exception
{
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
index 5760ea5fb10..02d523522c9 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
@@ -129,6 +129,24 @@ public class SessionVariablesTest extends
TestWithFeService {
Assertions.assertEquals(64, master.getQueryCacheEntryMaxRows());
}
+ @Test
+ public void testCloudGetVersionWaitForPendingTxn() throws Exception {
+ SessionVariable session = new SessionVariable();
+ SessionVariable forwarded = new SessionVariable();
+ Assertions.assertFalse(session.cloudGetVersionWaitForPendingTxn);
+ for (boolean waitForPendingTxn : new boolean[] {true, false}) {
+ VariableMgr.setVar(session, new SetVar(SetType.SESSION,
+ SessionVariable.CLOUD_GET_VERSION_WAIT_FOR_PENDING_TXN,
+ new StringLiteral(Boolean.toString(waitForPendingTxn))));
+ Assertions.assertEquals(waitForPendingTxn,
session.cloudGetVersionWaitForPendingTxn);
+ Map<String, String> vars = session.getForwardVariables();
+ Assertions.assertEquals(Boolean.toString(waitForPendingTxn),
+
vars.get(SessionVariable.CLOUD_GET_VERSION_WAIT_FOR_PENDING_TXN));
+ forwarded.setForwardedSessionVariables(vars);
+ Assertions.assertEquals(waitForPendingTxn,
forwarded.cloudGetVersionWaitForPendingTxn);
+ }
+ }
+
@Test
public void testInsertVisibleTimeoutReturnMode() throws Exception {
connectContext.setThreadLocalInfo();
diff --git a/gensrc/proto/cloud.proto b/gensrc/proto/cloud.proto
index 2e1463e5460..6d96eba78c0 100644
--- a/gensrc/proto/cloud.proto
+++ b/gensrc/proto/cloud.proto
@@ -1324,6 +1324,10 @@ message GetVersionResponse {
repeated int64 versions = 6;
repeated int64 version_update_time_ms = 7;
repeated int64 commit_tsos = 8;
+
+ // Pending state for partition versions, after any requested wait.
+ // One entry for scalar reads (including VERSION_NOT_FOUND), or per
partition in batch order.
+ repeated bool has_pending_txns = 9;
};
message GetObjStoreInfoRequest {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
