github-actions[bot] commented on code in PR #68104:
URL: https://github.com/apache/doris/pull/68104#discussion_r4032914010
##########
fe/fe-core/src/main/java/org/apache/doris/common/proc/PartitionsProcDir.java:
##########
@@ -422,7 +422,9 @@ private List<Long> getPartitionVersions(OlapTable
olapTable, List<Long> partitio
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));
+ // Lazy commit may advance the table version before its
partition versions become visible.
+
partitionVersions.addAll(CloudPartition.getSnapshotVisibleVersionFromMs(
+ batch, true,
Config.cloud_version_syncer_get_version_retry_times));
Review Comment:
[P1] Do not wait for pending transactions under metadata locks
This call runs while `getPartitionInfosInrernal` holds the table metadata
read lock and, for MTMVs, all related-table read locks. Setting
`waitForPendingTxns=true` turns a healthy pending lazy transaction into up to
three 30-second waits per sequential batch, blocking DDL for the entire
interval; the new unit test calls this helper outside the enclosing lock scope.
Snapshot partition identity under the metadata lock, release it for the waiting
RPCs, then reacquire and revalidate before publication.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java:
##########
@@ -164,48 +174,134 @@ private void syncPartitionVersion(Map<OlapTable, Long>
tableVersionMap) {
OlapTable olapTable = entry.getKey();
LOG.info("sync partition version for db: {}, table: {}, table
cache version: {}, new version: {}",
olapTable.getDatabase().getId(), olapTable,
olapTable.getCachedTableVersion(), entry.getValue());
- for (Partition partition : olapTable.getAllPartitions()) {
- partitions.add((CloudPartition) partition);
+ // Capture before fetching partitions; a later invalidation must
survive this round.
+ tableCacheEpochs.put(olapTable.getId(),
olapTable.getTableVersionCacheEpoch());
+ List<CloudPartition> currentPartitions;
+ olapTable.readLock();
+ try {
+ currentPartitions =
olapTable.getAllPartitions().stream().map(p -> (CloudPartition) p)
+ .collect(Collectors.toList());
+ } finally {
+ olapTable.readUnlock();
+ }
+ tablePartitions.put(olapTable.getId(), currentPartitions);
+ for (CloudPartition partition : currentPartitions) {
+ partitions.add(partition);
if (partitions.size() >=
Config.cloud_get_version_task_batch_size) {
- Future<Void> future =
submitGetPartitionVersionTask(failedTables, ImmutableList.copyOf(partitions));
+ Future<Void> future = submitGetPartitionVersionTask(
+ failedTables, versions,
ImmutableList.copyOf(partitions));
futures.add(future);
partitions.clear();
}
}
}
if (partitions.size() > 0) {
- Future<Void> future = submitGetPartitionVersionTask(failedTables,
ImmutableList.copyOf(partitions));
+ Future<Void> future = submitGetPartitionVersionTask(
+ failedTables, versions, ImmutableList.copyOf(partitions));
futures.add(future);
partitions.clear();
}
try {
for (Future<Void> future : futures) {
future.get();
}
- } catch (InterruptedException | ExecutionException e) {
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.error("Interrupted while waiting for get partition version
tasks to complete", e);
+ return;
+ } catch (ExecutionException e) {
LOG.error("Error waiting for get partition version tasks to
complete", e);
+ return;
}
- // set table version for success tables
+ // Separate RPC batches can straddle a commit. Publish only if the MS
table version stayed unchanged.
+ validateTableVersions(tableVersionMap, failedTables);
for (Entry<OlapTable, Long> entry : tableVersionMap.entrySet()) {
- if (!failedTables.contains(entry.getKey().getId())) {
- OlapTable olapTable = entry.getKey();
- olapTable.setCachedTableVersion(entry.getValue());
+ OlapTable olapTable = entry.getKey();
+ olapTable.readLock();
+ try {
+ olapTable.versionWriteLock();
+ try {
+ long cacheEpoch = tableCacheEpochs.get(olapTable.getId());
+ if (olapTable.getCachedTableVersion() > entry.getValue()
+ || olapTable.getTableVersionCacheEpoch() !=
cacheEpoch) {
+ failedTables.add(olapTable.getId());
+ }
+ if (!failedTables.contains(olapTable.getId())) {
+ // MS reads can advance a partition without updating
the table cache. Since partition
+ // setters never decrease versions, validate the whole
snapshot before applying any of it.
+ for (CloudPartition partition :
tablePartitions.get(olapTable.getId())) {
+ long version =
versions.get(partition.getId()).getVersion();
+ if (partition.getCachedVisibleVersion() > version)
{
+ failedTables.add(olapTable.getId());
+ LOG.info("partition cache advanced during
sync, table: {}, partition: {}, "
+ + "snapshot version: {},
cached version: {}", olapTable.getId(),
+ partition.getId(), version,
partition.getCachedVisibleVersion());
+ break;
+ }
+ }
+ }
+ if (failedTables.contains(olapTable.getId())) {
+ // Invalidate together so a query cannot combine a
failed batch with a successful one.
Review Comment:
[P2] Preserve newer pushed caches when a scan loses the race
Every newer token now schedules a full scan, including normal commits whose
partition updates were already installed coherently. If another such commit
advances the token during the scan, this branch rejects the staged snapshot and
invalidates all partition caches, including the newer coherent ones. A hot
table whose write cadence is shorter than the scan can repeat this every cycle
and force foreground MS refreshes. Discard the obsolete staged snapshot but
retain the newer caches and dirty marker/retry state, or distinguish complete
pushes from actual gaps.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java:
##########
@@ -155,7 +157,15 @@ private Future<Void>
submitGetTableVersionTask(Map<OlapTable, Long> tableVersion
}
private void syncPartitionVersion(Map<OlapTable, Long> tableVersionMap) {
+ // Keep retrying until every partition batch has completed, even if a
concurrent
+ // commit or table-version read refreshes the cache during this sync.
+ // Preserve the table cache until publication: SQL caches must not
label old partition data
+ // with a new table version fetched from MS while the partition
snapshot is still staged.
+
tableVersionMap.keySet().forEach(OlapTable::markPartitionVersionSyncNeeded);
Set<Long> failedTables = ConcurrentHashMap.newKeySet();
+ Map<Long, CloudPartition.PartitionVersion> versions = new
ConcurrentHashMap<>();
+ Map<Long, List<CloudPartition>> tablePartitions = new HashMap<>();
Review Comment:
[P2] Bound staged snapshots to one table
`tablePartitions`, task batches, and this catalog-wide map retain
references, map nodes, boxed IDs, and `PartitionVersion` objects for every
selected partition until every task and validation completes. The per-table
partition guard is mutable and the selected table count is unbounded, so a
round can create severe transient FE heap and GC pressure. Atomicity is
table-local; process, validate, publish, and release one table, or another
fixed-size bound, at a time.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudPartition.java:
##########
@@ -138,7 +138,9 @@ public void invalidateCachedVisibleVersion() {
@VisibleForTesting
protected boolean isCachedVersionExpired() {
- if (lastVersionCachedTimeMs == 0 || versionCacheEpoch.get() !=
refreshedVersionCacheEpoch.get()) {
+ // Without pushes or background sync, queries must refresh versions
themselves.
+ if (!Config.cloud_enable_version_syncer || lastVersionCachedTimeMs == 0
+ || versionCacheEpoch.get() !=
refreshedVersionCacheEpoch.get()) {
Review Comment:
[P1] Reconcile disabled-syncer behavior with finite TTLs
This leading condition bypasses every positive finite TTL whenever the
daemon is disabled. The unchanged two-FE
`test_fe_cached_partition_version.groovy` deliberately disables the syncer,
configures finite table/partition TTLs, and expects follower reads to remain
cached until expiry; this patch instead fetches current MS versions on each
read, so those assertions fail and RPC volume changes. If the safety
requirement is only for non-expiring caches, gate `Long.MAX_VALUE`; otherwise
update the existing regression and document the new config contract.
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudSyncVersionDaemon.java:
##########
@@ -80,6 +76,11 @@ private Map<OlapTable, Long> syncTableVersions() {
List<Long> dbIds = new ArrayList<>();
List<Long> tableIds = new ArrayList<>();
List<OlapTable> tables = new ArrayList<>();
+ // Finite TTLs refresh lazily on reads, but incomplete syncs must
still be retried
+ // after changing TTLs or re-enabling the daemon.
+ boolean syncExpiredTables =
VariableMgr.getDefaultSessionVariable().cloudPartitionVersionCacheTtlMs
+ == Long.MAX_VALUE
Review Comment:
[P1] Force a recovery sweep after re-enabling
If the final push is suppressed while the syncer is disabled and the
follower performs no read, its token remains old with no dirty marker. After
re-enable, finite defaults make `syncExpiredTables` false, so this selection
skips the table forever, while an already-open session with infinite TTLs can
keep serving the stale table and partitions. The new re-enable test reads while
disabled and thereby creates the marker. Track an enable generation and force
one sweep or invalidation on the transition; add the no-read missed-push case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
