This is an automated email from the ASF dual-hosted git repository.
CalvinKirs 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 7866f5a36c0 [fix](mtmv) Batch non-empty partition checks during MV
compensation (#69966)
7866f5a36c0 is described below
commit 7866f5a36c02e15dc014aaef30e537a395f4059d
Author: Calvin Kirs <[email protected]>
AuthorDate: Tue Nov 3 22:29:08 2026 +0800
[fix](mtmv) Batch non-empty partition checks during MV compensation (#69966)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
Cloud MV partition compensation checks each valid mapped MV partition
through a singleton non-empty request. Cached-empty or unknown
partitions can therefore cause many serial MS version requests, repeated
across related partition tables, even when the MV is not ultimately
selected.
Collect unique mapped valid MV partition IDs and call the existing bulk
interface in batches of at most 5,000. Reuse the returned ID set across
related tables, preserving partition removal, base-table compensation,
cache handling and exception propagation.
The 5,000-partition cap is a fixed code constant, not a runtime
configuration. It bounds the three request ID lists to at most 165,000
bytes at the worst protobuf integer width, plus the remaining request
fields. This is a request-size guard, not a claimed performance-optimal
batch size. Tests verify that 1,020 candidates use one bulk call and
10,001 use three; these are mocked interface call counts, not a live
Cloud benchmark.
---
.../rules/exploration/mv/PartitionCompensator.java | 27 ++-
.../mv/PartitionCompensatorBatchTest.java | 252 +++++++++++++++++++++
.../exploration/mv/PartitionCompensatorTest.java | 6 +-
3 files changed, 277 insertions(+), 8 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java
index 33636472c76..97b05fc3d4f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java
@@ -38,6 +38,7 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import org.apache.logging.log4j.LogManager;
@@ -52,6 +53,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.stream.Collectors;
/**
* Handle materialized view partition union compensate handler
@@ -65,6 +67,8 @@ public class PartitionCompensator {
// It means all partitions are used when query
public static final Collection<Pair<RelationId, Set<String>>>
ALL_PARTITIONS_LIST =
ImmutableList.of(ALL_PARTITIONS);
+ // Bound the number of partition versions requested by each non-empty
check.
+ private static final int NON_EMPTY_PARTITION_BATCH_SIZE = 5000;
/** Whether an async MV references a query relation with unavailable
partition coverage. */
public static boolean hasUnknownPartitionCoverage(MaterializationContext
materializationContext,
@@ -122,6 +126,18 @@ public class PartitionCompensator {
}
Map<MTMVRelatedTableIf, Map<String, Set<String>>>
mtmvRelatedTableIfMapMap
=
materializationContext.calculatePartitionMappings(queryUsedBaseTablePartitionMap);
+ Set<String> mappedMvPartitionNames = new HashSet<>();
+ mtmvRelatedTableIfMapMap.values().forEach(mapping ->
mappedMvPartitionNames.addAll(mapping.keySet()));
+ List<Long> mvPartitionIdsToCheck = mvValidPartitions.stream()
+ .filter(partition ->
mappedMvPartitionNames.contains(partition.getName()))
+ .map(Partition::getId)
+ .distinct()
+ .collect(Collectors.toList());
+ // An MV partition can be mapped to several PCT tables. Check it once
and reuse the result.
+ Set<Long> nonEmptyMvPartitionIds = new HashSet<>();
+ for (List<Long> batch : Lists.partition(mvPartitionIdsToCheck,
NON_EMPTY_PARTITION_BATCH_SIZE)) {
+
nonEmptyMvPartitionIds.addAll(mtmv.selectNonEmptyPartitionIds(batch,
Optional.empty()));
+ }
boolean allCompensateIsNull = true;
Map<BaseTableInfo, Set<String>> mvPartitionNeedRemoveNameMap = new
HashMap<>();
@@ -137,9 +153,10 @@ public class PartitionCompensator {
Set<String> relatedTableUsedPartitionSet
=
queryUsedBaseTablePartitionMap.get(relatedTable.getFullQualifiers());
Pair<Pair<BaseTableInfo, Set<String>>, Pair<BaseColInfo,
Set<String>>> needCompensatePartitions
- = getNeedCompensatePartitions(mvValidPartitions,
relatedTableUsedPartitionSet,
- rewrittenPlanUsePartitionNameSet, pctInfoMap.get(new
BaseTableInfo(relatedTable)),
- partitionMapping.getValue(), materializationContext);
+ = getNeedCompensatePartitions(mvValidPartitions,
nonEmptyMvPartitionIds,
+ relatedTableUsedPartitionSet,
rewrittenPlanUsePartitionNameSet,
+ pctInfoMap.get(new BaseTableInfo(relatedTable)),
partitionMapping.getValue(),
+ materializationContext);
allCompensateIsNull &= needCompensatePartitions == null;
if (needCompensatePartitions == null) {
continue;
@@ -178,6 +195,7 @@ public class PartitionCompensator {
private static Pair<Pair<BaseTableInfo, Set<String>>, Pair<BaseColInfo,
Set<String>>> getNeedCompensatePartitions(
Collection<Partition> mvValidPartitions,
+ Set<Long> nonEmptyMvPartitionIds,
Set<String> queryUsedBaseTablePartitionNameSet,
Set<String> rewrittenPlanUsePartitionNameSet,
BaseColInfo relatedPartitionTable,
@@ -196,8 +214,7 @@ public class PartitionCompensator {
Set<String> relatedBaseTablePartitions =
partitionMapping.get(mvValidPartition.getName());
if (relatedBaseTablePartitions != null) {
mvValidBaseTablePartitionNameSet.addAll(relatedBaseTablePartitions);
- if
(!mtmv.selectNonEmptyPartitionIds(ImmutableList.of(mvValidPartition.getId()),
- Optional.empty()).isEmpty()) {
+ if (nonEmptyMvPartitionIds.contains(mvValidPartition.getId()))
{
mvValidHasDataRelatedBaseTableNameSet.addAll(relatedBaseTablePartitions);
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorBatchTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorBatchTest.java
new file mode 100644
index 00000000000..ea839496586
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorBatchTest.java
@@ -0,0 +1,252 @@
+// 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.exploration.mv;
+
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.catalog.PartitionInfo;
+import org.apache.doris.catalog.PartitionType;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Pair;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.mtmv.BaseColInfo;
+import org.apache.doris.mtmv.BaseTableInfo;
+import org.apache.doris.mtmv.MTMVPartitionInfo;
+import org.apache.doris.mtmv.MTMVRelatedTableIf;
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.trees.plans.Plan;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+public class PartitionCompensatorBatchTest {
+ @ParameterizedTest
+ @ValueSource(ints = {1, 1020, 4999, 5000, 5001, 10001})
+ public void testBoundedBatchesAreSharedAcrossRelatedTables(int
partitionCount) throws Exception {
+ Fixture fixture = new Fixture();
+ for (long id = 1; id <= partitionCount; id++) {
+ fixture.addPartition(id);
+ }
+ // Duplicate candidates must not cause repeated version requests.
+ fixture.validPartitions.add(fixture.validPartitions.get(0));
+ fixture.queryUsedPartitions.replaceAll((table, names) ->
ImmutableSet.of("p1"));
+
+ List<List<Long>> requests = new ArrayList<>();
+
Mockito.when(fixture.mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any()))
+ .thenAnswer(invocation -> {
+ List<Long> ids =
ImmutableList.copyOf(invocation.<Collection<Long>>getArgument(0));
+ Assertions.assertEquals(Optional.empty(),
invocation.getArgument(1));
+ Assertions.assertTrue(ids.size() <= 5000);
+ requests.add(ids);
+ return ids;
+ });
+
+ Pair<Map<BaseTableInfo, Set<String>>, Map<BaseColInfo, Set<String>>>
result = fixture.calculate();
+ Assertions.assertNotNull(result);
+ Assertions.assertTrue(result.key().isEmpty());
+ Assertions.assertTrue(result.value().isEmpty());
+ Assertions.assertEquals((partitionCount + 4999) / 5000,
requests.size());
+ List<Long> requestedIds = new ArrayList<>();
+ requests.forEach(requestedIds::addAll);
+ Assertions.assertEquals(partitionCount, requestedIds.size());
+ Set<Long> uniqueRequestedIds = new HashSet<>(requestedIds);
+ Assertions.assertEquals(partitionCount, uniqueRequestedIds.size());
+ for (long id = 1; id <= partitionCount; id++) {
+ Assertions.assertTrue(uniqueRequestedIds.contains(id));
+ }
+ }
+
+ @Test
+ public void testMixedEmptyPartitionsPreserveCompensation() throws
Exception {
+ Fixture fixture = new Fixture();
+ fixture.addPartition(1L);
+ fixture.addPartition(2L);
+
fixture.queryUsedPartitions.put(fixture.firstTable.getFullQualifiers(),
ImmutableSet.of("p2", "t1_new"));
+
fixture.queryUsedPartitions.put(fixture.secondTable.getFullQualifiers(),
ImmutableSet.of("p2", "t2_new"));
+
Mockito.when(fixture.mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any()))
+ .thenReturn(ImmutableList.of(2L));
+
+ Pair<Map<BaseTableInfo, Set<String>>, Map<BaseColInfo, Set<String>>>
result = fixture.calculate();
+ Assertions.assertNotNull(result);
+ Assertions.assertTrue(result.key().isEmpty());
+ Assertions.assertEquals(2, result.value().size());
+ result.value().values().forEach(names ->
+ Assertions.assertEquals(ImmutableSet.of("t1_new", "t2_new"),
names));
+
Mockito.verify(fixture.mtmv).selectNonEmptyPartitionIds(ImmutableList.of(1L,
2L), Optional.empty());
+
+ // The only non-empty MV partition does not cover p1, so it cannot
answer this query.
+ fixture.queryUsedPartitions.replaceAll((table, names) ->
ImmutableSet.of("p1"));
+ Assertions.assertNull(fixture.calculate());
+ }
+
+ @Test
+ public void testAllEmptyPartitionsRejectMv() throws Exception {
+ Fixture fixture = new Fixture();
+ fixture.addPartition(1L);
+ fixture.addPartition(2L);
+
Mockito.when(fixture.mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any()))
+ .thenReturn(ImmutableList.of());
+
+ Assertions.assertNull(fixture.calculate());
+
Mockito.verify(fixture.mtmv).selectNonEmptyPartitionIds(ImmutableList.of(1L,
2L), Optional.empty());
+ }
+
+ @Test
+ public void testOnlyMappedValidPartitionsAreChecked() throws Exception {
+ Fixture fixture = new Fixture();
+ fixture.addPartition(1L);
+ fixture.addPartition(2L);
+ fixture.addPartition(3L);
+ fixture.validPartitions.remove(2);
+ fixture.partitionMappings.values().forEach(mapping ->
mapping.remove("mv_p2"));
+
Mockito.when(fixture.mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any()))
+ .thenAnswer(invocation ->
ImmutableList.copyOf(invocation.<Collection<Long>>getArgument(0)));
+
+ Assertions.assertNotNull(fixture.calculate());
+
Mockito.verify(fixture.mtmv).selectNonEmptyPartitionIds(ImmutableList.of(1L),
Optional.empty());
+ }
+
+ @Test
+ public void testNoMappedValidPartitionsSkipVersionRequests() throws
Exception {
+ Fixture fixture = new Fixture();
+ fixture.addPartition(1L);
+ fixture.partitionMappings.values().forEach(Map::clear);
+
+ Assertions.assertNull(fixture.calculate());
+ Mockito.verify(fixture.mtmv, Mockito.never())
+ .selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any());
+ }
+
+ @Test
+ public void testNoValidPartitionsSkipVersionRequests() throws Exception {
+ Fixture fixture = new Fixture();
+
+ Assertions.assertNull(fixture.calculate());
+ Mockito.verify(fixture.mtmv, Mockito.never())
+ .selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any());
+ }
+
+ @Test
+ public void testFailedBatchDoesNotReturnPartialCompensation() throws
Exception {
+ Fixture fixture = new Fixture();
+ for (long id = 1; id <= 5001; id++) {
+ fixture.addPartition(id);
+ }
+ RuntimeException failure = new RuntimeException("get version from meta
service failed");
+
Mockito.when(fixture.mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any()))
+ .thenAnswer(invocation ->
ImmutableList.copyOf(invocation.<Collection<Long>>getArgument(0)))
+ .thenThrow(failure);
+
+ Assertions.assertSame(failure,
Assertions.assertThrows(RuntimeException.class, fixture::calculate));
+ Mockito.verify(fixture.mtmv, Mockito.times(2))
+ .selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any());
+ }
+
+ private static class Fixture {
+ private final MTMV mtmv = Mockito.mock(MTMV.class);
+ private final List<Partition> validPartitions = new ArrayList<>();
+ private final Map<MTMVRelatedTableIf, Map<String, Set<String>>>
partitionMappings = new HashMap<>();
+ private final Map<List<String>, Set<String>> queryUsedPartitions = new
HashMap<>();
+ private final AsyncMaterializationContext materializationContext
+ = Mockito.mock(AsyncMaterializationContext.class);
+ private final CascadesContext cascadesContext =
Mockito.mock(CascadesContext.class);
+ private final Plan rewrittenPlan = Mockito.mock(Plan.class);
+ private final MTMVRelatedTableIf firstTable;
+ private final MTMVRelatedTableIf secondTable;
+
+ Fixture() throws AnalysisException {
+ CatalogIf<?> catalog = Mockito.mock(CatalogIf.class);
+ Mockito.when(catalog.getId()).thenReturn(1L);
+ Mockito.when(catalog.getName()).thenReturn("internal");
+ DatabaseIf<?> database = Mockito.mock(DatabaseIf.class);
+ Mockito.when(database.getId()).thenReturn(2L);
+ Mockito.when(database.getFullName()).thenReturn("db");
+ Mockito.when(database.getCatalog()).thenReturn(catalog);
+ firstTable = mockTable("t1", 10L, database);
+ secondTable = mockTable("t2", 20L, database);
+ partitionMappings.put(firstTable, new HashMap<>());
+ partitionMappings.put(secondTable, new HashMap<>());
+ queryUsedPartitions.put(firstTable.getFullQualifiers(),
ImmutableSet.of("p1"));
+ queryUsedPartitions.put(secondTable.getFullQualifiers(),
ImmutableSet.of("p1"));
+
+ Mockito.when(mtmv.getId()).thenReturn(100L);
+ Mockito.when(mtmv.getName()).thenReturn("mv");
+ Mockito.when(mtmv.getDatabase()).thenReturn(database);
+ PartitionInfo partitionInfo = Mockito.mock(PartitionInfo.class);
+
Mockito.when(partitionInfo.getType()).thenReturn(PartitionType.RANGE);
+ Mockito.when(mtmv.getPartitionInfo()).thenReturn(partitionInfo);
+ MTMVPartitionInfo mvPartitionInfo =
Mockito.mock(MTMVPartitionInfo.class);
+
Mockito.when(mvPartitionInfo.getPctTables()).thenReturn(ImmutableSet.of(firstTable,
secondTable));
+ List<BaseColInfo> pctInfos = ImmutableList.of(
+ new BaseColInfo("dt", new BaseTableInfo(firstTable)),
+ new BaseColInfo("dt", new BaseTableInfo(secondTable)));
+ Mockito.when(mvPartitionInfo.getPctInfos()).thenReturn(pctInfos);
+
Mockito.when(mtmv.getMvPartitionInfo()).thenReturn(mvPartitionInfo);
+
+ Mockito.when(materializationContext.getMtmv()).thenReturn(mtmv);
+
Mockito.when(materializationContext.calculatePartitionMappings(ArgumentMatchers.any()))
+ .thenReturn(partitionMappings);
+ StatementContext statementContext =
Mockito.mock(StatementContext.class);
+ Map<BaseTableInfo, Collection<Partition>> canRewritePartitions
+ = ImmutableMap.of(new BaseTableInfo(mtmv),
validPartitions);
+ Mockito.when(statementContext.getMvCanRewritePartitionsMap())
+ .thenReturn(canRewritePartitions);
+
Mockito.when(cascadesContext.getStatementContext()).thenReturn(statementContext);
+
Mockito.when(rewrittenPlan.collectToList(ArgumentMatchers.any())).thenReturn(ImmutableList.of());
+ }
+
+ private void addPartition(long id) {
+ String name = "mv_p" + id;
+ validPartitions.add(new Partition(id, name, null, null));
+ partitionMappings.values().forEach(mapping -> mapping.put(name,
ImmutableSet.of("p" + id)));
+ }
+
+ private Pair<Map<BaseTableInfo, Set<String>>, Map<BaseColInfo,
Set<String>>> calculate()
+ throws AnalysisException {
+ return PartitionCompensator.calcInvalidPartitions(
+ queryUsedPartitions, rewrittenPlan,
materializationContext, cascadesContext);
+ }
+
+ private static MTMVRelatedTableIf mockTable(String name, long id,
DatabaseIf<?> database) {
+ MTMVRelatedTableIf table = Mockito.mock(MTMVRelatedTableIf.class);
+ Mockito.when(table.getId()).thenReturn(id);
+ Mockito.when(table.getName()).thenReturn(name);
+ Mockito.when(table.getDatabase()).thenReturn(database);
+
Mockito.when(table.getFullQualifiers()).thenReturn(ImmutableList.of("internal",
"db", name));
+ return table;
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorTest.java
index b75a752bc8c..e6581055112 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensatorTest.java
@@ -455,7 +455,7 @@ public class PartitionCompensatorTest extends
TestWithFeService {
Mockito.when(mvPctInfo.getPctInfos()).thenReturn(ImmutableList.of(colInfo1,
colInfo2));
// All MV partitions contain data
Mockito.when(mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any()))
- .thenReturn(ImmutableList.of(1L));
+ .thenAnswer(invocation ->
ImmutableList.copyOf(invocation.<Collection<Long>>getArgument(0)));
AsyncMaterializationContext matCtx =
Mockito.mock(AsyncMaterializationContext.class);
Mockito.when(matCtx.getMtmv()).thenReturn(mtmv);
@@ -512,7 +512,7 @@ public class PartitionCompensatorTest extends
TestWithFeService {
Mockito.when(mtmv.getId()).thenReturn(100L);
Mockito.when(mtmv.getDatabase()).thenReturn(mvDb);
Mockito.when(mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any()))
- .thenReturn(ImmutableList.of(1L));
+ .thenAnswer(invocation ->
ImmutableList.copyOf(invocation.<Collection<Long>>getArgument(0)));
long validMvPartitionId = 101L;
long partiallyStaleMvPartitionId = 102L;
@@ -590,7 +590,7 @@ public class PartitionCompensatorTest extends
TestWithFeService {
Mockito.when(mtmv.getId()).thenReturn(100L);
Mockito.when(mtmv.getDatabase()).thenReturn(mvDb);
Mockito.when(mtmv.selectNonEmptyPartitionIds(ArgumentMatchers.any(),
ArgumentMatchers.any()))
- .thenReturn(ImmutableList.of(1L));
+ .thenAnswer(invocation ->
ImmutableList.copyOf(invocation.<Collection<Long>>getArgument(0)));
long mvP20260301Id = 101L;
long mvP20260401Id = 102L;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
