This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-incremental-computation
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-incremental-computation
by this push:
new 71679aca362 branch-incremental-computation: pick the merged
incremental-computation PRs from master in merge order (#68138) (#68303)
71679aca362 is described below
commit 71679aca3628ade1be3713c9f53ae30677a29db3
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Mon Sep 21 11:12:10 2026 +0800
branch-incremental-computation: pick the merged incremental-computation PRs
from master in merge order (#68138) (#68303)
Cherry-picked from #68138
Batch pick of every merged PR carrying the `incremental-computation`
label that `branch-incremental-computation` does not have yet (no
`incremental-computation-picked` label), in the order they landed on
master (`git log --first-parent`). One commit per PR, each ending with
`(cherry picked from commit <master sha>)`. Follows the same convention
as #67830, #68017, #68073, #68151 and #68236. This round has a single
PR.
| # | Master commit | PR | Title |
|---|---|---|---|
| 1 | 695c88b5772 | #68138 | [fix](binlog) Refresh incremental partition
versions before pruning |
Not included on purpose:
- The 28 labelled PRs that already carry
`incremental-computation-picked` (every other closed PR with the label).
### Prerequisite check
- **#68138** declares #67181 as related; #67181 (`e5a4e725fac`) is
before the fork point `efedf10c7e3`. Everything the pick uses already
exists on this branch:
`OlapTableWrapper.selectNonEmptyPartitionIds(Collection<Long>,
Optional<StreamReadMode>)` / `hasFixedVisibleVersions()`, the
two-argument `CloudPartition.getSnapshotVisibleVersionFromMs(List,
boolean)`, `StreamReadMode`, the three `RowBinlogTableWrapper`
constructors, and the test helpers
(`BinlogTestUtils.newTestRowBinlogConfig`,
`PlanConstructor.newOlapTable`, `MemoTestUtils.createConnectContext`,
`PlanChecker.from(ConnectContext, Plan)`). `cherry-pick -x` applied
cleanly (one auto-merge in `CloudPartition.java`, no conflict).
### Drift check against master
- `RowBinlogTableWrapper.java`, `OlapTableWrapperTest.java` and the new
`PruneEmptyPartitionTest.java` are byte-identical to master at
`695c88b5772`.
- `CloudPartition.java` differs from master only by the unlabelled
#66296 ("Reduce cloud version sync config": the `maxAttempts` overload
of `getSnapshotVisibleVersionFromMs` / `getSnapshotVisibleVersion` and
the `VariableMgr.getDefaultSessionVariable()` fallback for
`cloudPartitionVersionCacheTtlMs`). Applying #66296's hunks for this
file on top of the branch in a temporary index gives a zero-line diff
against master, so nothing of the pick is missing; the pick itself only
calls the two-argument `getSnapshotVisibleVersionFromMs(partitions,
false)`, which is the same on both sides. #66296 stays out, as in the
previous rounds.
### Verification
- FE: `run-fe-ut.sh --run` on this branch (regenerates thrift/protobuf,
compiles fe-core main + test) with the two touched test classes plus
every test class that exercises `selectNonEmptyPartitionIds` /
`getSnapshotVisibleVersionFromMs` / `RowBinlogTableWrapper`: 8 classes,
108 tests, 0 failures, 0 errors, BUILD SUCCESS —
`CloudGlobalTransactionMgrTest` 37, `ExplainTableStreamPlanTest` 24,
`PhysicalPlanTranslatorTest` 17, `OlapScanNodeTest` 12,
`PartitionCompensatorTest` 12, `OlapTableWrapperTest` 4 (the two new
tests included), `PruneEmptyPartitionTest` 1 (new),
`TableBinlogFunctionAuthTest` 1. The `@Test` counts of the two touched
classes equal master's.
- FE checkstyle on fe-core: 0 violations.
- No BE, cloud or regression-suite changes in this PR.
Co-authored-by: Luwei <[email protected]>
---
.../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 3dbd50e4ed0..d393bf347fa 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]
