This is an automated email from the ASF dual-hosted git repository.
luwei16 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 695c88b5772 [fix](binlog) Refresh incremental partition versions
before pruning (#68138)
695c88b5772 is described below
commit 695c88b57723ffd93c6a0978daa8725cb21b2562
Author: Luwei <[email protected]>
AuthorDate: Mon Sep 21 10:45:50 2026 +0800
[fix](binlog) Refresh incremental partition versions before pruning (#68138)
### What problem does this PR solve?
Issue Number: None
Related PR: #67181
Problem Summary:
A cloud incremental read may wait for its target transaction to become
visible before query planning continues. During that wait, the FE
`CloudPartition` cache can still contain `PARTITION_INIT_VERSION` for a
partition even though the transaction has already become visible in
MetaService.
`PruneEmptyPartition` previously relied only on that cached version. It
could therefore classify the partition as empty and remove it before
scan-node planning. The later version refresh in
`ScanNode.setVisibleVersionForOlapScanNodes` could observe the new
version, but it was already too late because the partition had been
pruned from the scan.
This PR changes empty-partition pruning for direct cloud incremental
reads as follows:
- A partition that is already known to be non-empty from the FE cache is
retained directly. Partition versions are monotonic, so a cached version
greater than `PARTITION_INIT_VERSION` remains sufficient for this
pruning decision.
- Only cached-empty or unknown candidate partitions are refreshed from
MetaService. This catches transactions that became visible after the FE
cache entry was populated while avoiding unnecessary RPC work for known
non-empty partitions.
- The refreshed and cached results are combined in the original
candidate-partition order.
- Normal reads and incremental reads backed by an `OlapTableWrapper`
with fixed visible versions continue to use their existing
snapshot-aware path.
The refresh in `ScanNode.setVisibleVersionForOlapScanNodes` is
intentionally retained. `PruneEmptyPartition` only needs an
empty-or-non-empty decision, while scan execution still needs the exact
snapshot versions of all final selected partitions.
### Release note
Fix cloud incremental reads that could miss newly visible data because a
stale FE partition-version cache caused premature empty-partition
pruning.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- `PruneEmptyPartitionTest`
- `OlapScanNodeTest`
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes. Direct cloud incremental reads refresh cached-empty or
unknown partition versions from MetaService before empty-partition
pruning.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../doris/catalog/RowBinlogTableWrapper.java | 23 +++++++
.../apache/doris/cloud/catalog/CloudPartition.java | 32 +++++++---
.../apache/doris/catalog/OlapTableWrapperTest.java | 67 ++++++++++++++++++++
.../rules/rewrite/PruneEmptyPartitionTest.java | 72 ++++++++++++++++++++++
4 files changed, 184 insertions(+), 10 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/RowBinlogTableWrapper.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/RowBinlogTableWrapper.java
index 3d88a5ffbf6..d10be2ef76e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/RowBinlogTableWrapper.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/RowBinlogTableWrapper.java
@@ -17,14 +17,21 @@
package org.apache.doris.catalog;
+import org.apache.doris.catalog.stream.StreamReadMode;
+import org.apache.doris.cloud.catalog.CloudPartition;
+import org.apache.doris.common.Config;
import org.apache.doris.common.Pair;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
+import java.util.Collection;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
/**
* A lightweight wrapper base for read binlog<Row> of table
@@ -79,6 +86,22 @@ public class RowBinlogTableWrapper extends OlapTableWrapper {
return KeysType.DUP_KEYS;
}
+ @Override
+ public List<Long> selectNonEmptyPartitionIds(Collection<Long> partitionIds,
+ Optional<StreamReadMode> streamReadMode) {
+ if (Config.isCloudMode() && !hasFixedVisibleVersions()) {
+ // A row-binlog scan can start immediately after its target
transaction becomes visible. Refresh
+ // cached-empty or unknown partitions so an older cache entry
cannot prune newly visible binlog data.
+ List<CloudPartition> partitions = partitionIds.stream()
+ .map(this::getPartition)
+ .filter(Objects::nonNull)
+ .map(partition -> (CloudPartition) partition)
+ .collect(Collectors.toList());
+ return CloudPartition.selectNonEmptyPartitionIdsFromMs(partitions);
+ }
+ return super.selectNonEmptyPartitionIds(partitionIds, streamReadMode);
+ }
+
@Override
public boolean equals(Object obj) {
if (!super.equals(obj)) {
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 07c2a6d504e..9c788a9ab7f 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
@@ -213,25 +213,37 @@ public class CloudPartition extends Partition {
// Select the non-empty partitions and return the ids.
public static List<Long> selectNonEmptyPartitionIds(List<CloudPartition>
partitions) {
- List<Long> nonEmptyPartitionIds = partitions.stream()
- .filter(CloudPartition::hasDataCached)
- .map(CloudPartition::getId)
- .collect(Collectors.toList());
- if (nonEmptyPartitionIds.size() == partitions.size()) {
+ return selectNonEmptyPartitionIds(partitions, false);
+ }
+
+ // Select non-empty partitions while bypassing the version cache for
cached-empty or unknown partitions.
+ public static List<Long>
selectNonEmptyPartitionIdsFromMs(List<CloudPartition> partitions) {
+ return selectNonEmptyPartitionIds(partitions, true);
+ }
+
+ private static List<Long> selectNonEmptyPartitionIds(List<CloudPartition>
partitions, boolean forceRefresh) {
+ List<Long> nonEmptyPartitionIds = new ArrayList<>(partitions.size());
+ List<CloudPartition> unknowns = new ArrayList<>(partitions.size());
+ for (CloudPartition partition : partitions) {
+ if (partition.hasDataCached()) {
+ nonEmptyPartitionIds.add(partition.getId());
+ } else {
+ unknowns.add(partition);
+ }
+ }
+ if (unknowns.isEmpty()) {
return nonEmptyPartitionIds;
}
- List<CloudPartition> unknowns = partitions.stream()
- .filter(p -> !p.hasDataCached())
- .collect(Collectors.toList());
-
SummaryProfile profile = getSummaryProfile();
if (profile != null) {
profile.incGetPartitionVersionByHasDataCount();
}
try {
- List<Long> versions =
CloudPartition.getSnapshotVisibleVersion(unknowns);
+ List<Long> versions = forceRefresh
+ ? CloudPartition.getSnapshotVisibleVersionFromMs(unknowns,
false)
+ : CloudPartition.getSnapshotVisibleVersion(unknowns);
int size = versions.size();
for (int i = 0; i < size; i++) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableWrapperTest.java
b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableWrapperTest.java
index 3d084f9e270..8c2c96ca4fc 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableWrapperTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/catalog/OlapTableWrapperTest.java
@@ -18,13 +18,21 @@
package org.apache.doris.catalog;
import org.apache.doris.binlog.BinlogTestUtils;
+import org.apache.doris.cloud.catalog.CloudPartition;
+import org.apache.doris.common.Config;
import org.apache.doris.thrift.TStorageType;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.TimeUnit;
public class OlapTableWrapperTest {
@@ -117,4 +125,63 @@ public class OlapTableWrapperTest {
Assertions.assertEquals(table.getSchemaByIndexId(table.getBaseIndexId()),
wrapper.getSchemaByIndexId(table.getBaseIndexId()));
Assertions.assertEquals(table.getIndexSchemaVersion(table.getBaseIndexId()),
wrapper.getIndexSchemaVersion(table.getBaseIndexId()));
}
+
+ @Test
+ public void testCloudRowBinlogWrapperRefreshesCachedEmptyPartitions() {
+ long stalePartitionId = 100L;
+ long cachedNonEmptyPartitionId = 101L;
+ List<Long> partitionIds = ImmutableList.of(stalePartitionId,
cachedNonEmptyPartitionId);
+
+ CloudPartition stalePartition = Mockito.mock(CloudPartition.class);
+ Mockito.when(stalePartition.getId()).thenReturn(stalePartitionId);
+ // Simulate another query refreshing the shared cache after the first
cached-state check.
+ Mockito.when(stalePartition.hasDataCached()).thenReturn(false, true);
+ CloudPartition cachedNonEmptyPartition =
Mockito.mock(CloudPartition.class);
+
Mockito.when(cachedNonEmptyPartition.getId()).thenReturn(cachedNonEmptyPartitionId);
+ Mockito.when(cachedNonEmptyPartition.hasDataCached()).thenReturn(true);
+
+ OlapTable table =
Mockito.spy(newTestTable(BinlogTestUtils.newTestRowBinlogConfig(true, true)));
+
Mockito.doReturn(stalePartition).when(table).getPartition(stalePartitionId);
+
Mockito.doReturn(cachedNonEmptyPartition).when(table).getPartition(cachedNonEmptyPartitionId);
+ RowBinlogTableWrapper wrapper = new RowBinlogTableWrapper(table);
+
+ try (MockedStatic<Config> mockedConfig =
Mockito.mockStatic(Config.class);
+ MockedStatic<CloudPartition> mockedPartition =
Mockito.mockStatic(
+ CloudPartition.class, Mockito.CALLS_REAL_METHODS)) {
+ mockedConfig.when(Config::isCloudMode).thenReturn(true);
+ mockedPartition.when(() ->
CloudPartition.getSnapshotVisibleVersionFromMs(
+ ImmutableList.of(stalePartition),
false)).thenReturn(ImmutableList.of(2L));
+ mockedPartition.clearInvocations();
+
+
Assertions.assertEquals(ImmutableList.of(cachedNonEmptyPartitionId,
stalePartitionId),
+ wrapper.selectNonEmptyPartitionIds(partitionIds,
Optional.empty()));
+ mockedPartition.verify(() ->
CloudPartition.getSnapshotVisibleVersionFromMs(
+ ImmutableList.of(stalePartition), false));
+ Mockito.verify(stalePartition).hasDataCached();
+ Mockito.verify(cachedNonEmptyPartition).hasDataCached();
+ }
+ }
+
+ @Test
+ public void
testCloudRowBinlogWrapperWithFixedVisibleVersionsUsesOriginTable() {
+ long emptyPartitionId = 100L;
+ long nonEmptyPartitionId = 101L;
+ List<Long> partitionIds = ImmutableList.of(emptyPartitionId,
nonEmptyPartitionId);
+
+ OlapTable table =
Mockito.spy(newTestTable(BinlogTestUtils.newTestRowBinlogConfig(true, true)));
+ Mockito.doReturn(ImmutableList.of(nonEmptyPartitionId)).when(table)
+ .selectNonEmptyPartitionIds(partitionIds, Optional.empty());
+ RowBinlogTableWrapper wrapper = new RowBinlogTableWrapper(table,
Collections.emptyMap(),
+ ImmutableMap.of(emptyPartitionId,
Partition.PARTITION_INIT_VERSION, nonEmptyPartitionId, 2L));
+
+ try (MockedStatic<Config> mockedConfig =
Mockito.mockStatic(Config.class);
+ MockedStatic<CloudPartition> mockedPartition =
Mockito.mockStatic(CloudPartition.class)) {
+ mockedConfig.when(Config::isCloudMode).thenReturn(true);
+
+ Assertions.assertEquals(ImmutableList.of(nonEmptyPartitionId),
+ wrapper.selectNonEmptyPartitionIds(partitionIds,
Optional.empty()));
+ Mockito.verify(table).selectNonEmptyPartitionIds(partitionIds,
Optional.empty());
+ mockedPartition.verifyNoInteractions();
+ }
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartitionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartitionTest.java
new file mode 100644
index 00000000000..819b2b54197
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/PruneEmptyPartitionTest.java
@@ -0,0 +1,72 @@
+// 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.nereids.rules.rewrite;
+
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.cloud.catalog.CloudPartition;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.util.MemoPatternMatchSupported;
+import org.apache.doris.nereids.util.MemoTestUtils;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.nereids.util.PlanConstructor;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.List;
+import java.util.Optional;
+
+class PruneEmptyPartitionTest implements MemoPatternMatchSupported {
+
+ @Test
+ void testNormalReadUsesCachedPartitionVersions() {
+ long emptyPartitionId = 100L;
+ long nonEmptyPartitionId = 101L;
+ List<Long> partitionIds = ImmutableList.of(emptyPartitionId,
nonEmptyPartitionId);
+
+ CloudPartition emptyPartition = Mockito.mock(CloudPartition.class);
+ Mockito.when(emptyPartition.getId()).thenReturn(emptyPartitionId);
+ CloudPartition nonEmptyPartition = Mockito.mock(CloudPartition.class);
+
Mockito.when(nonEmptyPartition.getId()).thenReturn(nonEmptyPartitionId);
+
+ OlapTable table = Mockito.spy(PlanConstructor.newOlapTable(10L,
"normal_tbl", 0));
+ Mockito.doReturn(partitionIds).when(table).getPartitionIds();
+
Mockito.doReturn(emptyPartition).when(table).getPartition(emptyPartitionId);
+
Mockito.doReturn(nonEmptyPartition).when(table).getPartition(nonEmptyPartitionId);
+ Mockito.doReturn(ImmutableList.of(nonEmptyPartitionId)).when(table)
+ .selectNonEmptyPartitionIds(Mockito.anyCollection(),
Mockito.any());
+
+ LogicalOlapScan scan = new LogicalOlapScan(
+ PlanConstructor.getNextRelationId(), table,
ImmutableList.of("normal_tbl"));
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+
+ try (MockedStatic<CloudPartition> mockedPartition =
Mockito.mockStatic(CloudPartition.class)) {
+ LogicalOlapScan rewritten = (LogicalOlapScan)
PlanChecker.from(connectContext, scan)
+ .applyTopDown(new PruneEmptyPartition())
+ .getPlan();
+
+ Assertions.assertEquals(ImmutableList.of(nonEmptyPartitionId),
rewritten.getSelectedPartitionIds());
+ Mockito.verify(table).selectNonEmptyPartitionIds(partitionIds,
Optional.empty());
+ mockedPartition.verifyNoInteractions();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
